You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by dorx <gi...@git.apache.org> on 2014/07/11 02:35:54 UTC
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
GitHub user dorx opened a pull request:
https://github.com/apache/spark/pull/1367
[SPARK-2359][MLlib] Correlations
Implementation for Pearson and Spearman's correlation.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/dorx/spark correlation
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/1367.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 #1367
----
commit dfc48548bf6e2f2f90ec4f5e47889db3dd842768
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-03T18:09:50Z
WIP
commit 2a40aa11e7f6b3ea5cb56bebc7cb797694e8aed5
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-04T01:10:39Z
initial, untested implementation of Pearson
commit 8e02c63e3a65c2b760ec8a6ccb38cf7414adc698
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-04T01:12:21Z
Merge branch 'master' into correlation
commit bb2e9776b915ce920446f9fbd8a731d09cb26f50
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-04T01:17:37Z
minor fix
commit 28561b6c8ef8ead5e323ea01d6fc98d556ad948a
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-06T20:19:56Z
wip
commit b843a13c49543add69b8604f653c82fa4cddeb5a
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-07T18:52:33Z
revert change in stat counter
commit 9d808ee86f30ba6bb9982b531fd5d683e3309108
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-09T22:21:03Z
wip units
commit 818fa315116339aa04c94407c66e09653d3eab7a
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-09T22:21:28Z
wip units
commit 534ebf297ecfea7f57c1d7de1d053f7fad02f992
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-09T22:22:35Z
Merge branch 'master' into correlation
commit b775ff966bf538c53e3da36ddd3dbfa8a2d1ac38
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-10T23:31:30Z
old wrong impl
commit bd2bacfd7c0a0b648f4cb51ef4718d8bffc510f7
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-11T00:20:27Z
Race condition bug
commit 63418847c757a515ba2caf0426b52eef498cc902
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-11T00:30:51Z
race condition bug squished
commit bd3cf19ee038a61040f2f234a5a5ec36e4e02a7d
Author: Doris Xin <do...@gmail.com>
Date: 2014-07-11T00:31:56Z
Merge branch 'master' into correlation
----
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-48682275
QA tests have started for PR 1367. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16536/consoleFull
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15036918
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
--- End diff --
RangePartitioner should be fine since getPartition works on distinct keys (so like you said, duplicate values are assigned to the same partition). As with many other things, we could just add a warning in the docs saying that if there's too many duplicates, use with caution.
On a related note, do you know how to detect if the item is the last when you're using flatMap on an iterator?
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-48953501
@mengxr Thanks for the feedback. Can you respond to my followup questions before I update my PR?
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855843
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/PearsonCorrelation.scala ---
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import breeze.linalg.{DenseMatrix => BDM}
+
+import org.apache.spark.mllib.linalg.{Matrices, Matrix, Vector}
+import org.apache.spark.mllib.linalg.distributed.RowMatrix
+import org.apache.spark.rdd.RDD
+
+/**
+ * Compute Pearson correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Pearson correlation can be found at
+ * http://en.wikipedia.org/wiki/Pearson_product-moment_correlation_coefficient
+ */
+object PearsonCorrelation extends Correlation {
--- End diff --
private[mllib] or [stat]
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15020475
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
+ val cogrouped = new CoGroupedRDD[Long](ranks, partitioner)
+ cogrouped.mapPartitions({ iter =>
--- End diff --
Please check my comment above about `preservePartitioning` before adding `preservePartitioning = true`.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808671
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/Statistics.scala ---
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat
+
+import org.apache.spark.mllib.linalg.{Matrix, Vector}
+import org.apache.spark.mllib.stat.correlation.Correlations
+import org.apache.spark.rdd.RDD
+
+object Statistics {
--- End diff --
Need doc for public classes. Please mark it as @Experimental for the initial version.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15028681
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
--- End diff --
number of partitions in input RDD is better given the two choices since defaultParallelism can be unset (in which case we need a copypasta of the second half of Partitioner.defaultPartitioner)
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855922
--- Diff: mllib/src/test/scala/org/apache/spark/mllib/stat/CorrelationSuite.scala ---
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat
+
+import org.scalatest.FunSuite
+
+import breeze.linalg.{DenseMatrix => BDM, Matrix => BM}
+
+import org.apache.spark.mllib.linalg.Vectors
+import org.apache.spark.mllib.stat.correlation.{Correlations, PearsonCorrelation,
+SpearmansCorrelation}
+import org.apache.spark.mllib.util.LocalSparkContext
+
+class CorrelationSuite extends FunSuite with LocalSparkContext {
+
+ // test input data
+ val xData = Array(1.0, 0.0, -2.0)
+ val yData = Array(4.0, 5.0, 3.0)
+ val data = Seq(
--- End diff --
Add a column with all `0`s or `1`s for 0 covariance.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15133828
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/Statistics.scala ---
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.mllib.linalg.{Matrix, Vector}
+import org.apache.spark.mllib.stat.correlation.Correlations
+import org.apache.spark.rdd.RDD
+
+/**
+ * API for statistical functions in MLlib
+ */
+@Experimental
+object Statistics {
+
+ /**
+ * Compute the Pearson correlation matrix for the input RDD of Vectors.
+ * Returns NaN if either vector has 0 variance.
+ *
+ * @param X an RDD[Vector] for which the correlation matrix is to be computed.
+ * @return Pearson correlation matrix comparing columns in X.
+ */
+ def corr(X: RDD[Vector]): Matrix = Correlations.corrMatrix(X)
+
+ /**
+ * Compute the correlation matrix for the input RDD of Vectors using the specified method.
+ * Methods currently supported: pearson (default), spearman
+ *
+ * Note that for Spearman, a rank correlation, we need to create an RDD[Double] for each column
+ * and sort it in order to retrieve the ranks and then join the columns back into an RDD[Vector],
+ * which is fairly costly. Cache the input RDD before calling corr with `method = "spearman"` to
+ * avoid recomputing the common lineage
--- End diff --
Missing `.` at the end.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808675
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/Statistics.scala ---
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat
+
+import org.apache.spark.mllib.linalg.{Matrix, Vector}
+import org.apache.spark.mllib.stat.correlation.Correlations
+import org.apache.spark.rdd.RDD
+
+object Statistics {
+
+ /**
+ * Compute the Pearson correlation matrix for the input RDD of Vectors.
+ */
+ def corr(X: RDD[Vector]): Matrix = Correlations.corrMatrix(X)
+
+ /**
+ * Compute the correlation matrix for the input RDD of Vectors using the specified method.
+ *
+ * Methods currently supported: pearson (default), spearman
--- End diff --
Follow ScalaDoc for methods. It is worth noting the cost for spearman.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855846
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/PearsonCorrelation.scala ---
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import breeze.linalg.{DenseMatrix => BDM}
+
+import org.apache.spark.mllib.linalg.{Matrices, Matrix, Vector}
+import org.apache.spark.mllib.linalg.distributed.RowMatrix
+import org.apache.spark.rdd.RDD
+
+/**
+ * Compute Pearson correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Pearson correlation can be found at
+ * http://en.wikipedia.org/wiki/Pearson_product-moment_correlation_coefficient
+ */
+object PearsonCorrelation extends Correlation {
+
+ /**
+ * Compute the Pearson correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute the Pearson correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val rowMatrix = new RowMatrix(X)
+ val cov = rowMatrix.computeCovariance()
+ computeCorrelationMatrixFromCovariance(cov)
+ }
+
+ /**
+ * Compute the pearson correlation matrix from the covariance matrix
+ */
+ def computeCorrelationMatrixFromCovariance(covarianceMatrix: Matrix): Matrix = {
+ val cov = covarianceMatrix.toBreeze.asInstanceOf[BDM[Double]]
+ val n = cov.cols
+
+ // Compute the standard deviation on the diagonals first
+ var i = 0
+ while (i < n) {
+ cov(i, i) = math.sqrt(cov(i, i))
+ i +=1
+ }
+ // or we could put the stddev in its own array to trade space for one less pass over the matrix
+
+ // TODO: use blas.dspr instead to compute the correlation matrix
+ // if the covariance matrix comes in the upper triangular form for free
+
+ // Loop through columns since cov is column major
+ var j = 0
+ var sigma = 0.0
+ while (j < n) {
+ sigma = cov(j, j)
+ i = 0
+ while (i < j) {
+ val covariance = cov(i, j) / (sigma * cov(i, i))
--- End diff --
`covariance` -> `corr`
Note that `cov(i, i)` could be zero.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15134018
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * Trait for correlation algorithms.
+ */
+private[stat] trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j. S(i, j) can be NaN if the correlation is undefined for column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]. Can be NaN if correlation is undefined for the
+ * input vectors.
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions { iter =>
+ iter.map { case (xi, yi) => new DenseVector(Array(xi, yi)) }
+ }
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Maintains the default correlation type, pearson
+ */
+private[stat] object Correlations {
+
+ // Note: after new types of correlations are implemented, please update this map
+ val nameToObjectMap = Map(("pearson", PearsonCorrelation), ("spearman", SpearmansCorrelation))
+ val defaultCorrName: String = "pearson"
+ val defaultCorr: Correlation = nameToObjectMap(defaultCorrName)
+
+ def corr(x: RDD[Double], y: RDD[Double], method: String = defaultCorrName): Double = {
+ val correlation = getCorrelationFromName(method)
+ correlation.computeCorrelation(x, y)
+ }
+
+ def corrMatrix(X: RDD[Vector], method: String = defaultCorrName): Matrix = {
+ val correlation = getCorrelationFromName(method)
+ correlation.computeCorrelationMatrix(X)
+ }
+
+ /**
+ * Perform simple string processing to match the input correlation name with a known name
+ *
+ * private to mllib for ease of unit testing
+ */
+ private[mllib] def getCorrelationFromName(method: String): Correlation = {
--- End diff --
This function could be further simplified since you have an object map. Querying on the map directly should be sufficient. Iterating over its entries is not necessary.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808705
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
--- End diff --
`({` -> ` {`
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-48682802
QA tests have started for PR 1367. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16537/consoleFull
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15134609
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/PearsonCorrelation.scala ---
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import breeze.linalg.{DenseMatrix => BDM}
+
+import org.apache.spark.Logging
+import org.apache.spark.mllib.linalg.{Matrices, Matrix, Vector}
+import org.apache.spark.mllib.linalg.distributed.RowMatrix
+import org.apache.spark.rdd.RDD
+
+/**
+ * Compute Pearson correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Pearson correlation can be found at
+ * http://en.wikipedia.org/wiki/Pearson_product-moment_correlation_coefficient
+ */
+private[stat] object PearsonCorrelation extends Correlation with Logging {
+
+ /**
+ * Compute the Pearson correlation for two datasets. NaN if either vector has 0 variance.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute the Pearson correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j. 0 covariance results in a correlation value of Double.NaN.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val rowMatrix = new RowMatrix(X)
+ val cov = rowMatrix.computeCovariance()
+ computeCorrelationMatrixFromCovariance(cov)
+ }
+
+ /**
+ * Compute the Pearson correlation matrix from the covariance matrix.
+ * 0 covariance results in a correlation value of Double.NaN.
+ */
+ def computeCorrelationMatrixFromCovariance(covarianceMatrix: Matrix): Matrix = {
+ val cov = covarianceMatrix.toBreeze.asInstanceOf[BDM[Double]]
+ val n = cov.cols
+
+ // Compute the standard deviation on the diagonals first
+ var i = 0
+ while (i < n) {
+ // TODO remove once covariance numerical issue resolved.
+ cov(i, i) = if (closeToZero(cov(i, i))) 0.0 else math.sqrt(cov(i, i))
+ i +=1
+ }
+
+ // Loop through columns since cov is column major
+ var j = 0
+ var sigma = 0.0
+ var containNaN = false
+ while (j < n) {
+ sigma = cov(j, j)
+ i = 0
+ while (i < j) {
+ val corr = if (sigma == 0.0 || cov(i, i) == 0.0) {
+ containNaN = true
+ Double.NaN
+ } else {
+ cov(i, j) / (sigma * cov(i, i))
+ }
+ cov(i, j) = corr
+ cov(j, i) = corr
+ i += 1
+ }
+ j += 1
+ }
+
+ // put 1.0 on the diagonals
+ i = 0
+ while (i < n) {
+ cov(i, i) = 1.0
+ i +=1
+ }
+
+ if (containNaN) {
+ logWarning("Pearson correlation matrix contains NaN values.")
+ }
+
+ Matrices.fromBreeze(cov)
+ }
+
+ private def closeToZero(value: Double, threshhold: Double = 10e-12): Boolean = {
+ math.abs(value - 0.0) <= threshhold
--- End diff --
`math.abs(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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855867
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
--- End diff --
Please add a warning if `numCols` is large, say, > 100. I'm not sure whether this is a good threshold. It would be great if you can try some examples and set a reasonable threshold.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15033708
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
--- End diff --
Hmm so one problem with the buffering approach for duplicates is that the uniqueIds for all the duplicate entries need to be saved in a list (since they're not necessarily sequential, so we can't do some kind of range thing). In the degenerate case where everything is a duplicate in a big RDD, this can be problematic (unless we want to get really fancy about saving ranges of IDs).
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-48682285
QA results for PR 1367:<br>- This patch FAILED unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>trait Correlation {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16536/consoleFull
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14897552
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/PearsonCorrelation.scala ---
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import breeze.linalg.{DenseMatrix => BDM}
+
+import org.apache.spark.mllib.linalg.{Matrices, Matrix, Vector}
+import org.apache.spark.mllib.linalg.distributed.RowMatrix
+import org.apache.spark.rdd.RDD
+
+/**
+ * Compute Pearson correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Pearson correlation can be found at
+ * http://en.wikipedia.org/wiki/Pearson_product-moment_correlation_coefficient
+ */
+object PearsonCorrelation extends Correlation {
+
+ /**
+ * Compute the Pearson correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute the Pearson correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val rowMatrix = new RowMatrix(X)
+ val cov = rowMatrix.computeCovariance()
+ computeCorrelationMatrixFromCovariance(cov)
+ }
+
+ /**
+ * Compute the pearson correlation matrix from the covariance matrix
+ */
+ def computeCorrelationMatrixFromCovariance(covarianceMatrix: Matrix): Matrix = {
+ val cov = covarianceMatrix.toBreeze.asInstanceOf[BDM[Double]]
+ val n = cov.cols
+
+ // Compute the standard deviation on the diagonals first
+ var i = 0
+ while (i < n) {
+ cov(i, i) = math.sqrt(cov(i, i))
+ i +=1
+ }
+ // or we could put the stddev in its own array to trade space for one less pass over the matrix
+
+ // TODO: use blas.dspr instead to compute the correlation matrix
+ // if the covariance matrix comes in the upper triangular form for free
+
+ // Loop through columns since cov is column major
+ var j = 0
+ var sigma = 0.0
+ while (j < n) {
+ sigma = cov(j, j)
+ i = 0
+ while (i < j) {
+ val covariance = cov(i, j) / (sigma * cov(i, i))
--- End diff --
What do you want returned when cov(i, i) is zero? Double.NaN? 0.0?
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15017975
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
--- End diff --
FWIW I tried it with zipWithUniqueId, and, as expected, the results were wrong when more than 1 partition is used.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15098446
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.HashPartitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ // TODO add warning for too many columns
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.mapPartitions[(Double, Long)] { iter =>
+ iter.map { case (vector, index) => (vector(k), index) }
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var rankSum = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
+ padded.flatMap { item =>
+ val rank = item._2 + 1 // zipWithIndex is 0-indexed but ranks are 1-indexed
--- End diff --
Added for the sake of consistency with the definition (also in case someone wants to pull the ranks function out to be used somewhere else). But I can remove it with a note in the docs saying this isn't verbatim with the def in the wiki link.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15134693
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.{Logging, HashPartitioner}
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+private[stat] object SpearmansCorrelation extends Correlation with Logging {
--- End diff --
Should it be called `SpearmanCorrelation` instead of `SpearmansCorrelation`?
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15019612
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
--- End diff --
Make sure you use `preservePartitioning` correctly. My first understanding was wrong and I may have used it wrongly somewhere. The argument should really be called `preservePartitioner`. It passes the partitioner to the derived RDD without checking. This may lead to wrong results. For example, the join in the following code block returns nothing, though `c` and `d` have the same content.
~~~
val a = sc.makeRDD(Seq((0, 1), (2, 3), (4, 5), (6, 7)), 4)
val b = a.groupByKey()
val c = b.mapPartitions({ iter =>
iter.flatMap { case (k, vv) =>
vv.map { v => (v, k) }
}}, true)
c.collect()
val d = sc.makeRDD(Seq((1, 0), (3, 2), (5, 4), (7, 6)), 4)
c.join(d).collect()
~~~
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15137023
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.{Logging, HashPartitioner}
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+private[stat] object SpearmansCorrelation extends Correlation with Logging {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ if (numCols > 50) {
+ logWarning("Computing the Spearman correlation matrix can be slow for large RDDs with more"
+ + " than 50 columns.")
+ }
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map { case (vector, index) => (vector(k), index) }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [2.5, 1.0, 0.0, 2.5]
+ * Note that positions here are 0-indexed, instead of the 1-indexed as in the definition for
+ * ranks in the standard definition for Spearman's correlation. This does not affect the final
+ * results and is slightly more performant.
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var firstRank = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
+ padded.flatMap { item =>
+ val rank = item._2
+ if (item._1._1 == lastVal && item._2 != Long.MinValue) {
+ IDBuffer += item._1._2
+ Iterator.empty
+ } else {
+ val entries = if (IDBuffer.size == 0) {
+ Iterator.empty
+ } else if (IDBuffer.size == 1) {
+ Iterator((IDBuffer(0), firstRank))
+ } else {
+ // averageRank = ((firstRank + IDBuffer.size) / 2.0) / IDBuffer.size
--- End diff --
I think this comment could be removed since the line below if not very hard to understand.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-49493613
QA results for PR 1367:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16837/consoleFull
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15098382
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.HashPartitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ // TODO add warning for too many columns
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.mapPartitions[(Double, Long)] { iter =>
+ iter.map { case (vector, index) => (vector(k), index) }
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var rankSum = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
+ padded.flatMap { item =>
+ val rank = item._2 + 1 // zipWithIndex is 0-indexed but ranks are 1-indexed
--- End diff --
It doesn't matter because it doesn't affect the variance and the correlation.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15099190
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.HashPartitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ // TODO add warning for too many columns
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.mapPartitions[(Double, Long)] { iter =>
+ iter.map { case (vector, index) => (vector(k), index) }
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var rankSum = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
+ padded.flatMap { item =>
+ val rank = item._2 + 1 // zipWithIndex is 0-indexed but ranks are 1-indexed
--- End diff --
Yes, please remove it and leave a note.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15033041
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
--- End diff --
`x.zip(y)` doesn't have a partitioner. `zip` preserves number of partitions and their indices but not the partitioner. That is why the argument name should be really called `preservePartitioner`. There is no reason to put `preservePartitioning` 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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14842983
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
--- End diff --
FWIW I tend to strongly agree that the algorithm should be specified by exactly one complete string. R isn't necessarily something to emulate. Even using Strings feels a bit old-school. I understand possible cross-platform concerns but what about passing in implementations of a Correlation interface, which are available as singleton instances for convenience from somewhere? Food for thought.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855906
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
--- End diff --
`ranks(i)` doesn't have a partitioner. So this should fall back to a default `HashPartitioner` with defaultParallelism. If that is the case, please use `HashPartitioner` directly to avoid confusion.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14896912
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
--- End diff --
Does that mean no toLowerCase either?
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855852
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
--- End diff --
Use `zipWithUniqueId`, which doesn't trigger a job.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15098472
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.HashPartitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ // TODO add warning for too many columns
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.mapPartitions[(Double, Long)] { iter =>
--- End diff --
Use `map` instead of `mapPartitions`.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15099185
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.HashPartitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ // TODO add warning for too many columns
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.mapPartitions[(Double, Long)] { iter =>
+ iter.map { case (vector, index) => (vector(k), index) }
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var rankSum = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
+ padded.flatMap { item =>
+ val rank = item._2 + 1 // zipWithIndex is 0-indexed but ranks are 1-indexed
+ if (item._1._1 == lastVal && item._2 != Long.MinValue) {
+ rankSum += rank
+ IDBuffer += item._1._2
+ Iterator.empty
+ } else {
+ val entries = if (IDBuffer.size == 0) {
+ Iterator.empty
+ } else if (IDBuffer.size == 1) {
+ Iterator((IDBuffer(0), rankSum))
+ } else {
+ IDBuffer.map(id => (id, rankSum / IDBuffer.size))
+ }
+ lastVal = item._1._1
+ rankSum = rank
+ IDBuffer.clear()
+ IDBuffer += item._1._2
+ entries
+ }
+ }
+ }
+ ranks
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]], input: RDD[Vector]): RDD[Vector] = {
+ val partitioner = new HashPartitioner(input.partitions.size)
+ val cogrouped = new CoGroupedRDD[Long](ranks, partitioner)
+ cogrouped.mapPartitions { iter =>
--- End diff --
I don't think there is performance difference. But `.map { ... }` is more concise than `.mapPartitions { iter => iter.map { ... } }`.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15019620
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
+ * supported method name has to be a substring of the input method name for it to be matched (e.g.
+ * "spearmansrho" will be matched to "spearman")
+ *
+ * Maintains the default correlation type, pearson
+ */
+object Correlations {
--- End diff --
Sounds good.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808684
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
--- End diff --
Trait for correlation algorithms.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-48688067
QA results for PR 1367:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds the following public classes (experimental):<br>trait Correlation {<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16537/consoleFull
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855845
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/PearsonCorrelation.scala ---
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import breeze.linalg.{DenseMatrix => BDM}
+
+import org.apache.spark.mllib.linalg.{Matrices, Matrix, Vector}
+import org.apache.spark.mllib.linalg.distributed.RowMatrix
+import org.apache.spark.rdd.RDD
+
+/**
+ * Compute Pearson correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Pearson correlation can be found at
+ * http://en.wikipedia.org/wiki/Pearson_product-moment_correlation_coefficient
+ */
+object PearsonCorrelation extends Correlation {
+
+ /**
+ * Compute the Pearson correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute the Pearson correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val rowMatrix = new RowMatrix(X)
+ val cov = rowMatrix.computeCovariance()
+ computeCorrelationMatrixFromCovariance(cov)
+ }
+
+ /**
+ * Compute the pearson correlation matrix from the covariance matrix
+ */
+ def computeCorrelationMatrixFromCovariance(covarianceMatrix: Matrix): Matrix = {
+ val cov = covarianceMatrix.toBreeze.asInstanceOf[BDM[Double]]
+ val n = cov.cols
+
+ // Compute the standard deviation on the diagonals first
+ var i = 0
+ while (i < n) {
+ cov(i, i) = math.sqrt(cov(i, i))
+ i +=1
+ }
+ // or we could put the stddev in its own array to trade space for one less pass over the matrix
+
+ // TODO: use blas.dspr instead to compute the correlation matrix
--- End diff --
dspr is for rank-1 symmetric update, which is irrelevant to the operation 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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855908
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
+ val cogrouped = new CoGroupedRDD[Long](ranks, partitioner)
+ cogrouped.mapPartitions({ iter =>
--- End diff --
replace `(` by a space
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15019889
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
--- End diff --
See my previous comment about mentioning the cost in the doc. User should be responsible to cache the RDD. If user already cached the input RDD, we are duplicating the data 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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15136854
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.{Logging, HashPartitioner}
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+private[stat] object SpearmansCorrelation extends Correlation with Logging {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ if (numCols > 50) {
+ logWarning("Computing the Spearman correlation matrix can be slow for large RDDs with more"
+ + " than 50 columns.")
+ }
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map { case (vector, index) => (vector(k), index) }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [2.5, 1.0, 0.0, 2.5]
+ * Note that positions here are 0-indexed, instead of the 1-indexed as in the definition for
+ * ranks in the standard definition for Spearman's correlation. This does not affect the final
+ * results and is slightly more performant.
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var firstRank = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
+ padded.flatMap { item =>
--- End diff --
Using `case ((v, id), rank) =>` here can improve the readability of the code block.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by srowen <gi...@git.apache.org>.
Github user srowen commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14898523
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/PearsonCorrelation.scala ---
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import breeze.linalg.{DenseMatrix => BDM}
+
+import org.apache.spark.mllib.linalg.{Matrices, Matrix, Vector}
+import org.apache.spark.mllib.linalg.distributed.RowMatrix
+import org.apache.spark.rdd.RDD
+
+/**
+ * Compute Pearson correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Pearson correlation can be found at
+ * http://en.wikipedia.org/wiki/Pearson_product-moment_correlation_coefficient
+ */
+object PearsonCorrelation extends Correlation {
+
+ /**
+ * Compute the Pearson correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute the Pearson correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val rowMatrix = new RowMatrix(X)
+ val cov = rowMatrix.computeCovariance()
+ computeCorrelationMatrixFromCovariance(cov)
+ }
+
+ /**
+ * Compute the pearson correlation matrix from the covariance matrix
+ */
+ def computeCorrelationMatrixFromCovariance(covarianceMatrix: Matrix): Matrix = {
+ val cov = covarianceMatrix.toBreeze.asInstanceOf[BDM[Double]]
+ val n = cov.cols
+
+ // Compute the standard deviation on the diagonals first
+ var i = 0
+ while (i < n) {
+ cov(i, i) = math.sqrt(cov(i, i))
+ i +=1
+ }
+ // or we could put the stddev in its own array to trade space for one less pass over the matrix
+
+ // TODO: use blas.dspr instead to compute the correlation matrix
+ // if the covariance matrix comes in the upper triangular form for free
+
+ // Loop through columns since cov is column major
+ var j = 0
+ var sigma = 0.0
+ while (j < n) {
+ sigma = cov(j, j)
+ i = 0
+ while (i < j) {
+ val covariance = cov(i, j) / (sigma * cov(i, i))
--- End diff --
I think the most honest result is NaN. R will return an error for example. You will get that already as the result of 0.0 / 0.0 in the JVM. It's worth documenting!
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15020400
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
--- End diff --
you're right. sorry i was thinking about the 2nd zipWithIndex call (on the sorted RDD). That one absolutely cannot be replaced with zipWithUnqueId or the results are all wrong.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14899116
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
--- End diff --
"Breaking ties" is a loose description of what happens. I actually need all items of the same value in the same partition in order to take the average of their positions in the sorted list. I'm open to suggestions on how to make it work with mapPartitions though.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15017783
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
--- End diff --
FWIW I tried it with zipWithUniqueId, and, as expected, the results were wrong when more than 1 partition is used.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15135411
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.{Logging, HashPartitioner}
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+private[stat] object SpearmansCorrelation extends Correlation with Logging {
--- End diff --
Spearman's correlation is a lot more prevalent (whereas Pearson correlation is much more common), but changing it to SpearmanCorrelation for consistency. This was why I wanted to provide fuzzy string matching.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14900429
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
+ val cogrouped = new CoGroupedRDD[Long](ranks, partitioner)
+ cogrouped.mapPartitions({ iter =>
--- End diff --
Missed the `preservesPartitioning = true`, which made the parenthesis necessary. Good catch.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15022756
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
--- End diff --
Thanks for the example explaining what `preservePartitioning` does. It seems like in this case it's a good idea to copy the partitioner into the pair RDD since zip preserves the partitioner, and if we don't preserve the partitioner here, we're unnecessarily dropping the partitioner even though we didn't need to shuffle the data at all.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808703
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
--- End diff --
We could make this the default implementation of `computeCorrelation(x, y)`.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15020198
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
--- End diff --
The `RangePartitioner` guarantees same keys appear in the same partition. Then inside each partition, you go through an iterator of (value: Double, rank: Long). Do not output a value and its rank directly. Count records with the same value until a different value comes up, then flush out the value and the average rank.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14836922
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
--- End diff --
I made it its own method on purpose to force future developers to think about whether this a good option or if custom code for vectors can be more performant.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14899055
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
--- End diff --
I need the exact index of each entry in order to compute accurate ranks in the case of duplicate values. Doesn't seem like zipWithUniqueId allows me to do that esp when duplicate values fall into different partitions. Suggestions on making it work with zipWithUniqueId?
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/1367
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15036690
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
--- End diff --
Right, we also need to maintain the list of original row ids. If a column only has one distinct value, groupBy may not work well either. We use Long ids. With a 100MB buffer, we can handle 12M entries. I hope it is already large enough for practical use. The groupBy operator may trigger a global shuffle, which is slow. Please check whether we can use the `RangePartitioner` obtained from `sortByKey` in groupBy so it doesn't trigger a global shuffle.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15098421
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.HashPartitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ // TODO add warning for too many columns
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.mapPartitions[(Double, Long)] { iter =>
+ iter.map { case (vector, index) => (vector(k), index) }
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var rankSum = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
+ padded.flatMap { item =>
+ val rank = item._2 + 1 // zipWithIndex is 0-indexed but ranks are 1-indexed
+ if (item._1._1 == lastVal && item._2 != Long.MinValue) {
+ rankSum += rank
+ IDBuffer += item._1._2
+ Iterator.empty
+ } else {
+ val entries = if (IDBuffer.size == 0) {
+ Iterator.empty
+ } else if (IDBuffer.size == 1) {
+ Iterator((IDBuffer(0), rankSum))
+ } else {
+ IDBuffer.map(id => (id, rankSum / IDBuffer.size))
+ }
+ lastVal = item._1._1
+ rankSum = rank
+ IDBuffer.clear()
+ IDBuffer += item._1._2
+ entries
+ }
+ }
+ }
+ ranks
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]], input: RDD[Vector]): RDD[Vector] = {
+ val partitioner = new HashPartitioner(input.partitions.size)
+ val cogrouped = new CoGroupedRDD[Long](ranks, partitioner)
+ cogrouped.mapPartitions { iter =>
--- End diff --
Use `map` instead of `mapPartitions`.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-49399247
QA tests have started for PR 1367. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16815/consoleFull
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14964217
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
--- End diff --
No, plain string matching should be sufficient.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15040598
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
--- End diff --
sounds good.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855912
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
+ val cogrouped = new CoGroupedRDD[Long](ranks, partitioner)
+ cogrouped.mapPartitions({ iter =>
+ iter.map {case (index, values:Seq[Seq[Double]]) => new DenseVector(values.flatten.toArray)}
--- End diff --
Add after `{`. If it hits the line width limit, move `DenseVector(values.flatten.toArray)` to the next line.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808804
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
--- End diff --
This feature actually makes it error-prone to add new correlations. For example, I use `corr(x, y, "p")` in my code and in v1.2 someone adds another correlation starting with "p". Then it breaks existing code. I saw R implements this feature but it simply limits the methods to pearson, spearman, and kendall. If they plan to add new correlation with prefix collision, they will face the same problem.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855844
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/PearsonCorrelation.scala ---
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import breeze.linalg.{DenseMatrix => BDM}
+
+import org.apache.spark.mllib.linalg.{Matrices, Matrix, Vector}
+import org.apache.spark.mllib.linalg.distributed.RowMatrix
+import org.apache.spark.rdd.RDD
+
+/**
+ * Compute Pearson correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Pearson correlation can be found at
+ * http://en.wikipedia.org/wiki/Pearson_product-moment_correlation_coefficient
+ */
+object PearsonCorrelation extends Correlation {
+
+ /**
+ * Compute the Pearson correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute the Pearson correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val rowMatrix = new RowMatrix(X)
+ val cov = rowMatrix.computeCovariance()
+ computeCorrelationMatrixFromCovariance(cov)
+ }
+
+ /**
+ * Compute the pearson correlation matrix from the covariance matrix
+ */
+ def computeCorrelationMatrixFromCovariance(covarianceMatrix: Matrix): Matrix = {
+ val cov = covarianceMatrix.toBreeze.asInstanceOf[BDM[Double]]
+ val n = cov.cols
+
+ // Compute the standard deviation on the diagonals first
+ var i = 0
+ while (i < n) {
+ cov(i, i) = math.sqrt(cov(i, i))
+ i +=1
+ }
+ // or we could put the stddev in its own array to trade space for one less pass over the matrix
--- End diff --
`cov` is local and has random access. Updating its diagonal is cheap.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808860
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
+ * supported method name has to be a substring of the input method name for it to be matched (e.g.
+ * "spearmansrho" will be matched to "spearman")
--- End diff --
This feature is not necessary.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-49487511
QA tests have started for PR 1367. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16837/consoleFull
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15033178
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
--- End diff --
The implementation of zip preserves the partitioner from the parent RDD. So it seems like for consistency sake the partitioner should be further preserved
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15019630
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/PearsonCorrelation.scala ---
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import breeze.linalg.{DenseMatrix => BDM}
+
+import org.apache.spark.mllib.linalg.{Matrices, Matrix, Vector}
+import org.apache.spark.mllib.linalg.distributed.RowMatrix
+import org.apache.spark.rdd.RDD
+
+/**
+ * Compute Pearson correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Pearson correlation can be found at
+ * http://en.wikipedia.org/wiki/Pearson_product-moment_correlation_coefficient
+ */
+object PearsonCorrelation extends Correlation {
+
+ /**
+ * Compute the Pearson correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute the Pearson correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val rowMatrix = new RowMatrix(X)
+ val cov = rowMatrix.computeCovariance()
+ computeCorrelationMatrixFromCovariance(cov)
+ }
+
+ /**
+ * Compute the pearson correlation matrix from the covariance matrix
+ */
+ def computeCorrelationMatrixFromCovariance(covarianceMatrix: Matrix): Matrix = {
+ val cov = covarianceMatrix.toBreeze.asInstanceOf[BDM[Double]]
+ val n = cov.cols
+
+ // Compute the standard deviation on the diagonals first
+ var i = 0
+ while (i < n) {
+ cov(i, i) = math.sqrt(cov(i, i))
+ i +=1
+ }
+ // or we could put the stddev in its own array to trade space for one less pass over the matrix
+
+ // TODO: use blas.dspr instead to compute the correlation matrix
+ // if the covariance matrix comes in the upper triangular form for free
+
+ // Loop through columns since cov is column major
+ var j = 0
+ var sigma = 0.0
+ while (j < n) {
+ sigma = cov(j, j)
+ i = 0
+ while (i < j) {
+ val covariance = cov(i, j) / (sigma * cov(i, i))
--- End diff --
Agree with @srowen : set NaN and generate a warning message. Btw, R returns a warning message instead of an error.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-48827327
@dorx Thanks for implementing correlations! I just made one pass. Besides the method name matching, please make sure only `Statistics.corr`s are public and the doc there are accurate.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855841
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
--- End diff --
It is not just for low maintenance cost. A strict contract also helps users understand exactly what to input and what to expect. For the fault tolerance, there are many ways to break Spark by putting an extra "s" somewhere. Let's not worry too much in this PR. If we find a mismatch at runtime, throw an error and list available methods.
Btw, I like git's solution. You cannot use "git commits" for "git commit" but git will suggest:
~~~
Did you mean this?
commit
~~~
Anyway, let's not spend time on this auto-suggestion feature either. Please update the PR with strict string matching. 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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855883
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
--- End diff --
Add doc explaining the return type.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15036448
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
--- End diff --
Did that before my last comment:
Here's zip:
def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)] = {
zipPartitions(other, true) { (thisIter, otherIter) =>
new Iterator[(T, U)] {
def hasNext = (thisIter.hasNext, otherIter.hasNext) match {
case (true, true) => true
case (false, false) => false
case _ => throw new SparkException("Can only zip RDDs with " +
"same number of elements in each partition")
}
def next = (thisIter.next, otherIter.next)
}
}
}
zipPartitions here returns new ZippedPartitionsRDD2(sc, sc.clean(f), this, rdd2, preservesPartitioning), which extends ZippedPartitionsBaseRDD, which has the following:
override val partitioner =
if (preservesPartitioning) firstParent[Any].partitioner else None
So yes, the partitioner is propagated from the parent via zip.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855879
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
--- End diff --
The code fits a single line: `val column = indexed.map { case (vector, index) => (vector(k), index) }`
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-49493898
LGTM. Thanks for implementing correlations! Merged into master.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14836617
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
--- End diff --
So the fact R has been supporting this feature for many years (and still haven't deprecated it) is good precedence that among the well known correlations, there aren't really any collisions. Since we require that only well known and widely adopted algorithms be added to mllib, the likelihood of collision is even smaller. (If someone's adding something in their own version of spark, presumably they have already looked closely at the docs and been warned).
Because we decided to go with strings for method specification instead of enums for cross language uniformity, the method is only checked (and possibly invalidated) at run time. Therefore, we do want to provide mechanisms for fault tolerance and minimization in method specification. To address a related comment, "spearman" and "spearmans" are both really common in referring to Spearman's correlation (imagine the heartbreak of having your program fail because of an extra "s" after hours of data manipulation 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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15098384
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.HashPartitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ // TODO add warning for too many columns
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.mapPartitions[(Double, Long)] { iter =>
+ iter.map { case (vector, index) => (vector(k), index) }
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var rankSum = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
+ padded.flatMap { item =>
+ val rank = item._2 + 1 // zipWithIndex is 0-indexed but ranks are 1-indexed
+ if (item._1._1 == lastVal && item._2 != Long.MinValue) {
+ rankSum += rank
--- End diff --
Storing the first rank is sufficient. The average rank can be derived from first rank and `IDBuffer.size`:
~~~
firstRank + IDBuffer.size / 2.0
~~~
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-49467775
QA tests have started for PR 1367. This patch merges cleanly. <br>View progress: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16830/consoleFull
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855891
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
--- End diff --
`sortByKey` is done through a range partitioner, which means records having the same key will be inside the same partition (please check the code and confirm). So `mapPartitions` should be sufficient to break ties.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14855861
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
--- End diff --
If `zipWithUnqueId` is used, this block becomes unnecessary.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14896742
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
--- End diff --
No can do here since I have a second argument (preservePartitioning)
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-48682999
@mengrx, @falaki, @jkbradley please take a look
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15030155
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
+ val cogrouped = new CoGroupedRDD[Long](ranks, partitioner)
+ cogrouped.mapPartitions({ iter =>
--- End diff --
pretty sure we want to preserve the partitioner here (o/w it's gets wiped in the map operation).
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15040646
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
+ val cogrouped = new CoGroupedRDD[Long](ranks, partitioner)
+ cogrouped.mapPartitions({ iter =>
--- End diff --
The output is sent to `Pearson.computeCorrelationMatrix`, which uses covariance to derive the correlation matrix. No `groupBy` or `join` is used. The partitioner has no function here and it shouldn't be carried over because the output is `RDD[Vector]` but the partitioner applies to `index`, which no longer exists in the data.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14897906
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
--- End diff --
you would still branch the lineage into numCols RDDs, which then get joined back into a single RDD. Does the joining back prevent recalculation of the common lineage prefix when it's branched?
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15133829
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/Statistics.scala ---
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.mllib.linalg.{Matrix, Vector}
+import org.apache.spark.mllib.stat.correlation.Correlations
+import org.apache.spark.rdd.RDD
+
+/**
+ * API for statistical functions in MLlib
+ */
+@Experimental
+object Statistics {
+
+ /**
+ * Compute the Pearson correlation matrix for the input RDD of Vectors.
+ * Returns NaN if either vector has 0 variance.
+ *
+ * @param X an RDD[Vector] for which the correlation matrix is to be computed.
+ * @return Pearson correlation matrix comparing columns in X.
+ */
+ def corr(X: RDD[Vector]): Matrix = Correlations.corrMatrix(X)
+
+ /**
+ * Compute the correlation matrix for the input RDD of Vectors using the specified method.
+ * Methods currently supported: pearson (default), spearman
+ *
+ * Note that for Spearman, a rank correlation, we need to create an RDD[Double] for each column
+ * and sort it in order to retrieve the ranks and then join the columns back into an RDD[Vector],
+ * which is fairly costly. Cache the input RDD before calling corr with `method = "spearman"` to
+ * avoid recomputing the common lineage
+ *
+ * @param X an RDD[Vector] for which the correlation matrix is to be computed.
+ * @param method String specifying the method to use for computing correlation
--- End diff --
Besides mentioning in the method doc, could you also list the supported method names and the default values in this line? It may be easier for users to find what the options are.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-49477683
QA results for PR 1367:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16830/consoleFull
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/1367#issuecomment-49405645
QA results for PR 1367:<br>- This patch PASSES unit tests.<br>- This patch merges cleanly<br>- This patch adds no public classes<br><br>For more information see test ouptut:<br>https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/16815/consoleFull
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15135115
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.{Logging, HashPartitioner}
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+private[stat] object SpearmansCorrelation extends Correlation with Logging {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ if (numCols > 50) {
+ logWarning("Computing the Spearman correlation matrix can be slow for large RDDs with more"
+ + " than 50 columns.")
+ }
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map { case (vector, index) => (vector(k), index) }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [2.5, 1.0, 0.0, 2.5]
+ * Note that positions here are 0-indexed, instead of the 1-indexed as in the definition for
+ * ranks in the standard definition for Spearman's correlation. This does not affect the final
+ * results and is slightly more performant.
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var firstRank = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
--- End diff --
Use camelCase for normal variables. We did use capital letters for matrices in MLlib but not for normal Java objects.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15133921
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * Trait for correlation algorithms.
+ */
+private[stat] trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j. S(i, j) can be NaN if the correlation is undefined for column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]. Can be NaN if correlation is undefined for the
+ * input vectors.
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions { iter =>
--- End diff --
`map` is more concise here than `mapPartitions`
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808707
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
--- End diff --
spaces around `case` and add one space before `}`
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14900326
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
--- End diff --
That reminds me, changed `val column` to be created using `mapPartitions` instead in order to preserve original partitioning in order to avoid shuffling, so they should have partitioner inherited from `indexed`. In either case, I think it's actually cleaner to use the Partitioner.defaultPartitioner since it's exactly intended to create a partitions given a bunch of RDDs. (It does one more check if default.parallelism isn't set, and I don't think there's point in copying in the second half of the function since it's just bloating the codebase with duplicate logic.)
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14836715
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
+ * supported method name has to be a substring of the input method name for it to be matched (e.g.
+ * "spearmansrho" will be matched to "spearman")
+ *
+ * Maintains the default correlation type, pearson
+ */
+object Correlations {
--- End diff --
The reason I named it Correlations isn't to copy Vectors but to signify the fact that this is more of a delegator object that maintains all supported correlations.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15134590
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/PearsonCorrelation.scala ---
@@ -0,0 +1,107 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import breeze.linalg.{DenseMatrix => BDM}
+
+import org.apache.spark.Logging
+import org.apache.spark.mllib.linalg.{Matrices, Matrix, Vector}
+import org.apache.spark.mllib.linalg.distributed.RowMatrix
+import org.apache.spark.rdd.RDD
+
+/**
+ * Compute Pearson correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Pearson correlation can be found at
+ * http://en.wikipedia.org/wiki/Pearson_product-moment_correlation_coefficient
+ */
+private[stat] object PearsonCorrelation extends Correlation with Logging {
+
+ /**
+ * Compute the Pearson correlation for two datasets. NaN if either vector has 0 variance.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute the Pearson correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j. 0 covariance results in a correlation value of Double.NaN.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val rowMatrix = new RowMatrix(X)
+ val cov = rowMatrix.computeCovariance()
+ computeCorrelationMatrixFromCovariance(cov)
+ }
+
+ /**
+ * Compute the Pearson correlation matrix from the covariance matrix.
+ * 0 covariance results in a correlation value of Double.NaN.
+ */
+ def computeCorrelationMatrixFromCovariance(covarianceMatrix: Matrix): Matrix = {
+ val cov = covarianceMatrix.toBreeze.asInstanceOf[BDM[Double]]
+ val n = cov.cols
+
+ // Compute the standard deviation on the diagonals first
+ var i = 0
+ while (i < n) {
+ // TODO remove once covariance numerical issue resolved.
+ cov(i, i) = if (closeToZero(cov(i, i))) 0.0 else math.sqrt(cov(i, i))
+ i +=1
+ }
+
+ // Loop through columns since cov is column major
+ var j = 0
+ var sigma = 0.0
+ var containNaN = false
+ while (j < n) {
+ sigma = cov(j, j)
+ i = 0
+ while (i < j) {
+ val corr = if (sigma == 0.0 || cov(i, i) == 0.0) {
+ containNaN = true
+ Double.NaN
+ } else {
+ cov(i, j) / (sigma * cov(i, i))
+ }
+ cov(i, j) = corr
+ cov(j, i) = corr
+ i += 1
+ }
+ j += 1
+ }
+
+ // put 1.0 on the diagonals
+ i = 0
+ while (i < n) {
+ cov(i, i) = 1.0
+ i +=1
+ }
+
+ if (containNaN) {
+ logWarning("Pearson correlation matrix contains NaN values.")
+ }
+
+ Matrices.fromBreeze(cov)
+ }
+
+ private def closeToZero(value: Double, threshhold: Double = 10e-12): Boolean = {
--- End diff --
Using `10e-12` is not very common. Did you mean `1e-12` or `1e-11`?
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808688
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
--- End diff --
`private[stat]`?
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15019606
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
--- End diff --
It is okay to keep it as long as the `Correlation` class is private.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808679
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/Statistics.scala ---
@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat
+
+import org.apache.spark.mllib.linalg.{Matrix, Vector}
+import org.apache.spark.mllib.stat.correlation.Correlations
+import org.apache.spark.rdd.RDD
+
+object Statistics {
+
+ /**
+ * Compute the Pearson correlation matrix for the input RDD of Vectors.
+ */
+ def corr(X: RDD[Vector]): Matrix = Correlations.corrMatrix(X)
+
+ /**
+ * Compute the correlation matrix for the input RDD of Vectors using the specified method.
+ *
+ * Methods currently supported: pearson (default), spearman
+ */
+ def corr(X: RDD[Vector], method: String): Matrix = Correlations.corrMatrix(X, method)
+
+ /**
+ * Compute the Pearson correlation for the input RDDs.
+ */
+ def corr(x: RDD[Double], y: RDD[Double]): Double = Correlations.corr(x, y)
+
+ /**
+ * Compute the correlation for the input RDDs using the specified method.
+ *
+ * Methods currently supported: pearson (default), spearman
+ */
+ def corr(x: RDD[Double], y: RDD[Double], method: String): Double = Correlations.corr(x, y, method)
+
--- End diff --
remove empty lines
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808714
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
+ * supported method name has to be a substring of the input method name for it to be matched (e.g.
+ * "spearmansrho" will be matched to "spearman")
+ *
+ * Maintains the default correlation type, pearson
+ */
+object Correlations {
--- End diff --
Change to `Correlation` and mark it `private[stat]`. We used `Vectors` because `scala.Vector` is imported by default.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15040517
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
--- End diff --
Sorry, I forgot that I was on branch-0.9 today ... Even `zip` carries the partitioner, I don't understand why the partitioner should be preserved in our case. The output is `RDD[Vector]` but the partitioner only applies to its first element. This is error-prone. The final join is on unique row ids, irreverent to the partitioner in `x`. Please check this partitioner is correctly used somewhere or we should remove it.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15135084
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,128 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.{Logging, HashPartitioner}
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+private[stat] object SpearmansCorrelation extends Correlation with Logging {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ if (numCols > 50) {
+ logWarning("Computing the Spearman correlation matrix can be slow for large RDDs with more"
+ + " than 50 columns.")
+ }
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map { case (vector, index) => (vector(k), index) }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [2.5, 1.0, 0.0, 2.5]
+ * Note that positions here are 0-indexed, instead of the 1-indexed as in the definition for
+ * ranks in the standard definition for Spearman's correlation. This does not affect the final
+ * results and is slightly more performant.
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var firstRank = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
+ padded.flatMap { item =>
+ val rank = item._2
+ if (item._1._1 == lastVal && item._2 != Long.MinValue) {
+ IDBuffer += item._1._2
+ Iterator.empty
+ } else {
+ val entries = if (IDBuffer.size == 0) {
+ Iterator.empty
+ } else if (IDBuffer.size == 1) {
+ Iterator((IDBuffer(0), firstRank))
+ } else {
+ // averageRank = ((firstRank + IDBuffer.size) / 2.0) / IDBuffer.size
+ val averageRank = firstRank / (2 * IDBuffer.size) + 0.5
--- End diff --
Should be `firstRank + (IDBuffer.size - 1) / 2.0`. I was wrong in my previous comment.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14842782
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
--- End diff --
R only supports 3 correlations and they happen to not have name collisions, which is fine because there was no plan to support more correlations in R on a single machine. But if there comes a popular correlation called "kuma", the way R can handle it is renaming it to some name like "zuma" or explaining to user why "ku" is required for "kuma" but "k" is for "kendall". That was a bad decision R's developers made.
The implementation in this PR suggests that developers can add their own implementation but the name resolution isn't really expandable. Besides kendall's tau, we might add approximate algorithms in the future, which could have name like "approx_spearman" and "approx_pearson". Then we are facing collisions in the first character or non-unique substring matching.
Also, we want to make the behavior consistent in MLlib. It adds burden to future development. The best solution to me is exact string matching, simple implementation and low maintenance cost.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15020402
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
+ // Attempt to checkpoint the RDD before splitting it into numCols RDD[Double]s to avoid
+ // computing the lineage prefix multiple times.
+ // If checkpoint directory not set, cache the RDD instead.
+ try {
+ indexed.checkpoint()
+ } catch {
+ case e: Exception => indexed.cache()
+ }
+
+ val numCols = X.first.size
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.map {case(vector, index) => {
+ (vector(k), index)}
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+ val groupedByValue = sorted.groupBy(_._1._1)
+ val ranks = groupedByValue.flatMap[(Long, Double)] { item =>
+ val duplicates = item._2
+ if (duplicates.size > 1) {
+ val averageRank = duplicates.foldLeft(0L) {_ + _._2 + 1} / duplicates.size.toDouble
+ duplicates.map(entry => (entry._1._2, averageRank)).toSeq
+ } else {
+ duplicates.map(entry => (entry._1._2, entry._2.toDouble + 1)).toSeq
+ }
+ }
+ ranks.sortByKey()
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]]): RDD[Vector] = {
+ val partitioner = Partitioner.defaultPartitioner(ranks(0), ranks.tail: _*)
--- End diff --
The code is weird because `ranks` do share any partitioner. It is cleaner to use HashPartitioner with the same number of partitions in the input RDD or simply `defaultParallelisim`.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15019642
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.Partitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithIndex()
--- End diff --
The input `X` is not ordered. I think the purpose here is just to assign each row an id for joining back the elements after we obtain the rank for each column.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by jkbradley <gi...@git.apache.org>.
Github user jkbradley commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15087547
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
+ * supported method name has to be a substring of the input method name for it to be matched (e.g.
+ * "spearmansrho" will be matched to "spearman")
+ *
+ * Maintains the default correlation type, pearson
+ */
+object Correlations {
+
+ // Note: after new types of correlations are implemented, please update this map
+ val nameToObjectMap = Map(("pearson", PearsonCorrelation), ("spearman", SpearmansCorrelation))
+ val defaultCorrName: String = "pearson"
+ val defaultCorr: Correlation = nameToObjectMap(defaultCorrName)
+
+ def corr(x: RDD[Double], y: RDD[Double], method: String = defaultCorrName): Double = {
+ val correlation = getCorrelationFromName(method)
+ correlation.computeCorrelation(x, y)
+ }
+
+ def corrMatrix(X: RDD[Vector], method: String = defaultCorrName): Matrix = {
+ val correlation = getCorrelationFromName(method)
+ correlation.computeCorrelationMatrix(X)
+ }
+
+ /**
+ * Perform simple string processing to match the input correlation name with a known name
+ *
+ * private to mllib for ease of unit testing
+ */
+ private[mllib] def getCorrelationFromName(method: String): Correlation = {
+ if (method.equals(defaultCorrName)) {
+ defaultCorr
+ } else {
+ var correlation: Correlation = defaultCorr
+ var matched = false
+ val initialsAllowed = areInitialsUnique()
+ val inputLower = method.toLowerCase()
+ nameToObjectMap.foreach { case (name, corr) =>
+ if (!matched) {
+ if ((initialsAllowed && method.size == 1 && name.startsWith(inputLower)) ||
+ inputLower.contains(name)) { // match names like "spearmans"
--- End diff --
Style issue: Line wrap & spacing
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15133872
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/Statistics.scala ---
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat
+
+import org.apache.spark.annotation.Experimental
+import org.apache.spark.mllib.linalg.{Matrix, Vector}
+import org.apache.spark.mllib.stat.correlation.Correlations
+import org.apache.spark.rdd.RDD
+
+/**
+ * API for statistical functions in MLlib
+ */
+@Experimental
+object Statistics {
+
+ /**
+ * Compute the Pearson correlation matrix for the input RDD of Vectors.
+ * Returns NaN if either vector has 0 variance.
+ *
+ * @param X an RDD[Vector] for which the correlation matrix is to be computed.
+ * @return Pearson correlation matrix comparing columns in X.
+ */
+ def corr(X: RDD[Vector]): Matrix = Correlations.corrMatrix(X)
+
+ /**
+ * Compute the correlation matrix for the input RDD of Vectors using the specified method.
+ * Methods currently supported: pearson (default), spearman
+ *
+ * Note that for Spearman, a rank correlation, we need to create an RDD[Double] for each column
+ * and sort it in order to retrieve the ranks and then join the columns back into an RDD[Vector],
+ * which is fairly costly. Cache the input RDD before calling corr with `method = "spearman"` to
+ * avoid recomputing the common lineage
+ *
+ * @param X an RDD[Vector] for which the correlation matrix is to be computed.
+ * @param method String specifying the method to use for computing correlation
+ * @return Correlation matrix comparing columns in X.
+ */
+ def corr(X: RDD[Vector], method: String): Matrix = Correlations.corrMatrix(X, method)
+
+ /**
+ * Compute the Pearson correlation for the input RDDs.
+ * Columns with 0 covariance produce NaN entries in the correlation matrix.
+ *
+ * @param x RDD[Double] of the same cardinality as y
+ * @param y RDD[Double] of the same cardinality as x
+ * @return A Double containing the Pearson correlation between the two input RDD[Double]s
+ */
+ def corr(x: RDD[Double], y: RDD[Double]): Double = Correlations.corr(x, y)
+
+ /**
+ * Compute the correlation for the input RDDs using the specified method.
+ * Methods currently supported: pearson (default), spearman
+ *
+ * @param x RDD[Double] of the same cardinality as y
+ * @param y RDD[Double] of the same cardinality as x
+ * @param method String specifying the method to use for computing correlation
--- 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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14846509
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
+ iter.map {case(xi, yi) => new DenseVector(Array(xi, yi))}
+ }, preservesPartitioning = true)
+ computeCorrelationMatrix(mat)(0, 1)
+ }
+
+}
+
+/**
+ * Delegates computation to the specific correlation object based on the input method name
+ *
+ * Currently supported correlations: pearson, spearman.
+ * After new correlation algorithms are added, please update the documentation here and in
+ * Statistics.scala for the correlation APIs.
+ *
+ * Cases are ignored when doing method matching. We also allow initials, e.g. "P" for "pearson", as
+ * long as initials are unique in the supported set of correlation algorithms. In addition, a
--- End diff --
While I agree we definitely should anticipate cases for generalization, I feel ambivalent about generalizing based on fairly hypothetical use cases (especially since there hasn't been issues of name collision in well known correlation algorithms for many years in almost all major stats packages).
In any case, maybe fuzzy string matching isn't the optimal solution here, but the consideration was around user friendliness and the fact that the method name isn't something that gets checked at compile time (unless we find a mechanism to do so). Since Spark attracts users with its fault tolerance and user friendliness, it seems silly to me to have something fail at runtime (potentially after a lot of other data processing computation) because of an extra "s" in "spearmans"(and both "spearman" and "spearmans" seem popular options). The Correlation interface idea sounds interesting, and I can see people who need things to be fault tolerant going the extra mile of implementing it (but of course by default we don't require it and use exact string matching instead).
On the other hand, I understand that developers can only do so much to tolerate fault in user behaviors, and maintainability seems much more important in comparison.
At any rate, if it comes down to exact string matching or no deal, I'll gladly go with exact string matching only.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r14808250
--- Diff: python/pyspark/join.py ---
@@ -1,35 +1,19 @@
-"""
-Copyright (c) 2011, Douban Inc. <http://www.douban.com/>
-All rights reserved.
-
-Redistribution and use in source and binary forms, with or without
-modification, are permitted provided that the following conditions are
-met:
-
- * Redistributions of source code must retain the above copyright
-notice, this list of conditions and the following disclaimer.
-
- * Redistributions in binary form must reproduce the above
-copyright notice, this list of conditions and the following disclaimer
-in the documentation and/or other materials provided with the
-distribution.
-
- * Neither the name of the Douban Inc. nor the names of its
-contributors may be used to endorse or promote products derived from
-this software without specific prior written permission.
-
-THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
-"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
-LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
-A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
-OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
-SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
-LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
-DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
-THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
-(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
-OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-"""
--- End diff --
This was discussed in https://github.com/apache/spark/pull/592 , and the conclusion was to keep the Douban BSD license. Also, this is irrelevant to this PR. So please remove this change.
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by dorx <gi...@git.apache.org>.
Github user dorx commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15098523
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/SpearmansCorrelation.scala ---
@@ -0,0 +1,125 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import scala.collection.mutable.ArrayBuffer
+
+import org.apache.spark.HashPartitioner
+import org.apache.spark.SparkContext._
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.{CoGroupedRDD, RDD}
+
+/**
+ * Compute Spearman's correlation for two RDDs of the type RDD[Double] or the correlation matrix
+ * for an RDD of the type RDD[Vector].
+ *
+ * Definition of Spearman's correlation can be found at
+ * http://en.wikipedia.org/wiki/Spearman's_rank_correlation_coefficient
+ */
+object SpearmansCorrelation extends Correlation {
+
+ /**
+ * Compute Spearman's correlation for two datasets.
+ */
+ override def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double = {
+ computeCorrelationWithMatrixImpl(x, y)
+ }
+
+ /**
+ * Compute Spearman's correlation matrix S, for the input matrix, where S(i, j) is the
+ * correlation between column i and j.
+ *
+ * Input RDD[Vector] should be cached or checkpointed if possible since it would be split into
+ * numCol RDD[Double]s, each of which sorted, and the joined back into a single RDD[Vector].
+ */
+ override def computeCorrelationMatrix(X: RDD[Vector]): Matrix = {
+ val indexed = X.zipWithUniqueId()
+
+ val numCols = X.first.size
+ // TODO add warning for too many columns
+ val ranks = new Array[RDD[(Long, Double)]](numCols)
+
+ // Note: we use a for loop here instead of a while loop with a single index variable
+ // to avoid race condition caused by closure serialization
+ for (k <- 0 until numCols) {
+ val column = indexed.mapPartitions[(Double, Long)] { iter =>
+ iter.map { case (vector, index) => (vector(k), index) }
+ }
+ ranks(k) = getRanks(column)
+ }
+
+ val ranksMat: RDD[Vector] = makeRankMatrix(ranks, X)
+ PearsonCorrelation.computeCorrelationMatrix(ranksMat)
+ }
+
+ /**
+ * Compute the ranks for elements in the input RDD, using the average method for ties.
+ *
+ * With the average method, elements with the same value receive the same rank that's computed
+ * by taking the average of their positions in the sorted list.
+ * e.g. ranks([2, 1, 0, 2]) = [3.5, 2.0, 1.0, 3.5]
+ *
+ * @param indexed RDD[(Double, Long)] containing pairs of the format (originalValue, uniqueId)
+ * @return RDD[(Long, Double)] containing pairs of the format (uniqueId, rank), where uniqueId is
+ * copied from the input RDD.
+ */
+ private def getRanks(indexed: RDD[(Double, Long)]): RDD[(Long, Double)] = {
+ // Get elements' positions in the sorted list for computing average rank for duplicate values
+ val sorted = indexed.sortByKey().zipWithIndex()
+
+ val ranks: RDD[(Long, Double)] = sorted.mapPartitions { iter =>
+ // add an extra element to signify the end of the list so that flatMap can flush the last
+ // batch of duplicates
+ val padded = iter ++
+ Iterator[((Double, Long), Long)](((Double.NaN, Long.MinValue), Long.MinValue))
+ var lastVal = 0.0
+ var rankSum = 0.0
+ val IDBuffer = new ArrayBuffer[Long]()
+ padded.flatMap { item =>
+ val rank = item._2 + 1 // zipWithIndex is 0-indexed but ranks are 1-indexed
+ if (item._1._1 == lastVal && item._2 != Long.MinValue) {
+ rankSum += rank
+ IDBuffer += item._1._2
+ Iterator.empty
+ } else {
+ val entries = if (IDBuffer.size == 0) {
+ Iterator.empty
+ } else if (IDBuffer.size == 1) {
+ Iterator((IDBuffer(0), rankSum))
+ } else {
+ IDBuffer.map(id => (id, rankSum / IDBuffer.size))
+ }
+ lastVal = item._1._1
+ rankSum = rank
+ IDBuffer.clear()
+ IDBuffer += item._1._2
+ entries
+ }
+ }
+ }
+ ranks
+ }
+
+ private def makeRankMatrix(ranks: Array[RDD[(Long, Double)]], input: RDD[Vector]): RDD[Vector] = {
+ val partitioner = new HashPartitioner(input.partitions.size)
+ val cogrouped = new CoGroupedRDD[Long](ranks, partitioner)
+ cogrouped.mapPartitions { iter =>
--- End diff --
Care to edify me on the performance difference? Smaller closure to be serialized? What's a good rule of thumb for when to use `map` v `mapPartitions`?
---
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.
---
[GitHub] spark pull request: [SPARK-2359][MLlib] Correlations
Posted by mengxr <gi...@git.apache.org>.
Github user mengxr commented on a diff in the pull request:
https://github.com/apache/spark/pull/1367#discussion_r15036312
--- Diff: mllib/src/main/scala/org/apache/spark/mllib/stat/correlation/Correlation.scala ---
@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.mllib.stat.correlation
+
+import org.apache.spark.mllib.linalg.{DenseVector, Matrix, Vector}
+import org.apache.spark.rdd.RDD
+
+/**
+ * New correlation algorithms should implement this trait
+ */
+trait Correlation {
+
+ /**
+ * Compute correlation for two datasets.
+ */
+ def computeCorrelation(x: RDD[Double], y: RDD[Double]): Double
+
+ /**
+ * Compute the correlation matrix S, for the input matrix, where S(i, j) is the correlation
+ * between column i and j.
+ */
+ def computeCorrelationMatrix(X: RDD[Vector]): Matrix
+
+ /**
+ * Combine the two input RDD[Double]s into an RDD[Vector] and compute the correlation using the
+ * correlation implementation for RDD[Vector]
+ */
+ def computeCorrelationWithMatrixImpl(x: RDD[Double], y: RDD[Double]): Double = {
+ val mat: RDD[Vector] = x.zip(y).mapPartitions({ iter =>
--- End diff --
Could you double check and confirm? `x.zip(y)` should return a `ZippedRDD`, which doesn't implement `partitioner` and hence it falls back to the default one: `None`.
---
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.
---