diff options
author | z001qdp <Nicholas.Eggert@target.com> | 2016-07-15 12:30:22 +0100 |
---|---|---|
committer | Sean Owen <sowen@cloudera.com> | 2016-07-15 12:30:22 +0100 |
commit | 71ad945bbbdd154eae852cd7f841e98f7a83e8d4 (patch) | |
tree | 9d6d5b62dba642b46978a729a968e0057faecaf8 /mllib/src/main | |
parent | 1832423827fd518853b63f91c321e4568a39107d (diff) | |
download | spark-71ad945bbbdd154eae852cd7f841e98f7a83e8d4.tar.gz spark-71ad945bbbdd154eae852cd7f841e98f7a83e8d4.tar.bz2 spark-71ad945bbbdd154eae852cd7f841e98f7a83e8d4.zip |
[SPARK-16426][MLLIB] Fix bug that caused NaNs in IsotonicRegression
## What changes were proposed in this pull request?
Fixed a bug that caused `NaN`s in `IsotonicRegression`. The problem occurs when training rows with the same feature value but different labels end up on different partitions. This patch changes a `sortBy` call to a `partitionBy(RangePartitioner)` followed by a `mapPartitions(sortBy)` in order to ensure that all rows with the same feature value end up on the same partition.
## How was this patch tested?
Added a unit test.
Author: z001qdp <Nicholas.Eggert@target.com>
Closes #14140 from neggert/SPARK-16426-isotonic-nan.
Diffstat (limited to 'mllib/src/main')
-rw-r--r-- | mllib/src/main/scala/org/apache/spark/mllib/regression/IsotonicRegression.scala | 9 |
1 files changed, 6 insertions, 3 deletions
diff --git a/mllib/src/main/scala/org/apache/spark/mllib/regression/IsotonicRegression.scala b/mllib/src/main/scala/org/apache/spark/mllib/regression/IsotonicRegression.scala index 1cd6f2a896..377326f873 100644 --- a/mllib/src/main/scala/org/apache/spark/mllib/regression/IsotonicRegression.scala +++ b/mllib/src/main/scala/org/apache/spark/mllib/regression/IsotonicRegression.scala @@ -35,6 +35,7 @@ import org.apache.spark.mllib.linalg.{Vector, Vectors} import org.apache.spark.mllib.util.{Loader, Saveable} import org.apache.spark.rdd.RDD import org.apache.spark.sql.SparkSession +import org.apache.spark.RangePartitioner /** * Regression model for isotonic regression. @@ -408,9 +409,11 @@ class IsotonicRegression private (private var isotonic: Boolean) extends Seriali */ private def parallelPoolAdjacentViolators( input: RDD[(Double, Double, Double)]): Array[(Double, Double, Double)] = { - val parallelStepResult = input - .sortBy(x => (x._2, x._1)) - .glom() + val keyedInput = input.keyBy(_._2) + val parallelStepResult = keyedInput + .partitionBy(new RangePartitioner(keyedInput.getNumPartitions, keyedInput)) + .values + .mapPartitions(p => Iterator(p.toArray.sortBy(x => (x._2, x._1)))) .flatMap(poolAdjacentViolators) .collect() .sortBy(x => (x._2, x._1)) // Sort again because collect() doesn't promise ordering. |