add a new test to sliding

Author: mengxr

mllib/src/main/scala/org/apache/spark/mllib/evaluation/AreaUnderCurve.scala

@@ -18,15 +18,22 @@
 package org.apache.spark.mllib.evaluation
 
 import org.apache.spark.rdd.RDD
+import org.apache.spark.mllib.rdd.RDDFunctions._
 
 /**
  * Computes the area under the curve (AUC) using the trapezoidal rule.
  */
 object AreaUnderCurve {
 
-  private def trapezoid(points: Array[(Double, Double)]): Double = {
+  /**
+   * Uses the trapezoidal rule to compute the area under the line connecting the two input points.
+   * @param points two 2D points stored in Seq
+   */
+  private def trapezoid(points: Seq[(Double, Double)]): Double = {
     require(points.length == 2)
-    (points(1)._1 - points(0)._1) * (points(1)._2 + points(0)._2 ) / 2.0
+    val x = points.head
+    val y = points.last
+    (y._1 - x._1) * (y._2 + x._2) / 2.0
   }
 
   /**
@@ -36,20 +43,20 @@ object AreaUnderCurve {
    */
   def of(curve: RDD[(Double, Double)]): Double = {
     curve.sliding(2).aggregate(0.0)(
-      seqOp = (auc: Double, points: Array[(Double, Double)]) => auc + trapezoid(points),
-      combOp = (_ + _)
+      seqOp = (auc: Double, points: Seq[(Double, Double)]) => auc + trapezoid(points),
+      combOp = _ + _
     )
   }
 
   /**
    * Returns the area under the given curve.
    *
-   * @param curve an iterable of ordered 2D points stored in pairs representing a curve
+   * @param curve an iterator over ordered 2D points stored in pairs representing a curve
    */
-  def of(curve: Iterable[(Double, Double)]): Double = {
-    curve.sliding(2).map(_.toArray).filter(_.size == 2).aggregate(0.0)(
-      seqop = (auc: Double, points: Array[(Double, Double)]) => auc + trapezoid(points),
-      combop = (_ + _)
+  def of(curve: Iterator[(Double, Double)]): Double = {
+    curve.sliding(2).withPartial(false).aggregate(0.0)(
+      seqop = (auc: Double, points: Seq[(Double, Double)]) => auc + trapezoid(points),
+      combop = _ + _
     )
   }
 }

mllib/src/test/scala/org/apache/spark/mllib/evaluation/AreaUnderCurveSuite.scala

@@ -26,21 +26,21 @@ class AreaUnderCurveSuite extends FunSuite with LocalSparkContext {
   test("auc computation") {
     val curve = Seq((0.0, 0.0), (1.0, 1.0), (2.0, 3.0), (3.0, 0.0))
     val auc = 4.0
-    assert(AreaUnderCurve.of(curve) === auc)
+    assert(AreaUnderCurve.of(curve.toIterator) === auc)
     val rddCurve = sc.parallelize(curve, 2)
     assert(AreaUnderCurve.of(rddCurve) == auc)
   }
 
   test("auc of an empty curve") {
     val curve = Seq.empty[(Double, Double)]
-    assert(AreaUnderCurve.of(curve) === 0.0)
+    assert(AreaUnderCurve.of(curve.toIterator) === 0.0)
     val rddCurve = sc.parallelize(curve, 2)
     assert(AreaUnderCurve.of(rddCurve) === 0.0)
   }
 
   test("auc of a curve with a single point") {
     val curve = Seq((1.0, 1.0))
-    assert(AreaUnderCurve.of(curve) === 0.0)
+    assert(AreaUnderCurve.of(curve.toIterator) === 0.0)
     val rddCurve = sc.parallelize(curve, 2)
     assert(AreaUnderCurve.of(rddCurve) === 0.0)
   }

mllib/src/test/scala/org/apache/spark/mllib/rdd/RDDFunctionsSuite.scala

@@ -29,12 +29,21 @@ class RDDFunctionsSuite extends FunSuite with LocalSparkContext {
     for (numPartitions <- 1 to 8) {
       val rdd = sc.parallelize(data, numPartitions)
       for (windowSize <- 1 to 6) {
-        val slided = rdd.sliding(windowSize).collect().map(_.toList).toList
+        val sliding = rdd.sliding(windowSize).collect().map(_.toList).toList
         val expected = data.sliding(windowSize).map(_.toList).toList
-        assert(slided === expected)
+        assert(sliding === expected)
       }
       assert(rdd.sliding(7).collect().isEmpty,
         "Should return an empty RDD if the window size is greater than the number of items.")
     }
   }
+
+  test("sliding with empty partitions") {
+    val data = Seq(Seq(1, 2, 3), Seq.empty[Int], Seq(4), Seq.empty[Int], Seq(5, 6, 7))
+    val rdd = sc.parallelize(data, data.length).flatMap(s => s)
+    assert(rdd.partitions.size === data.length)
+    val sliding = rdd.sliding(3)
+    val expected = data.flatMap(x => x).sliding(3).toList
+    assert(sliding.collect().toList === expected)
+  }
 }