Add more comments, made all PID constant parameters positive, a couple more tests.

Author: dragos

streaming/src/main/scala/org/apache/spark/streaming/scheduler/rate/PIDRateEstimator.scala

@@ -28,20 +28,22 @@ package org.apache.spark.streaming.scheduler.rate
  *
  * @param batchDurationMillis the batch duration, in milliseconds
  * @param proportional how much the correction should depend on the current
- *        error. This term usually provides the bulk of correction. A value too large would
- *        make the controller overshoot the setpoint, while a small value would make the
- *        controller too insensitive. The default value is -1.
+ *        error. This term usually provides the bulk of correction and should be positive or zero.
+ *        A value too large would make the controller overshoot the setpoint, while a small value
+ *        would make the controller too insensitive. The default value is 1.
  * @param integral how much the correction should depend on the accumulation
- *        of past errors. This term accelerates the movement towards the setpoint, but a large
- *        value may lead to overshooting. The default value is -0.2.
+ *        of past errors. This value should be positive or 0. This term accelerates the movement
+ *        towards the desired value, but a large value may lead to overshooting. The default value
+ *        is 0.2.
  * @param derivative how much the correction should depend on a prediction
- *        of future errors, based on current rate of change. This term is not used very often,
- *        as it impacts stability of the system. The default value is 0.
+ *        of future errors, based on current rate of change. This value should be positive or 0.
+ *        This term is not used very often, as it impacts stability of the system. The default
+ *        value is 0.
  */
 private[streaming] class PIDRateEstimator(
     batchIntervalMillis: Long,
-    proportional: Double = -1D,
-    integral: Double = -.2D,
+    proportional: Double = 1D,
+    integral: Double = .2D,
     derivative: Double = 0D)
   extends RateEstimator {
 
@@ -53,9 +55,19 @@ private[streaming] class PIDRateEstimator(
   require(
     batchIntervalMillis > 0,
     s"Specified batch interval $batchIntervalMillis in PIDRateEstimator is invalid.")
+  require(
+    proportional >= 0,
+    s"Proportional term $proportional in PIDRateEstimator should be >= 0.")
+  require(
+    integral >= 0,
+    s"Integral term $integral in PIDRateEstimator should be >= 0.")
+  require(
+    derivative >= 0,
+    s"Derivative term $derivative in PIDRateEstimator should be >= 0.")
+
 
   def compute(time: Long, // in milliseconds
-      elements: Long,
+      numElements: Long,
       processingDelay: Long, // in milliseconds
       schedulingDelay: Long // in milliseconds
     ): Option[Double] = {
@@ -67,16 +79,19 @@ private[streaming] class PIDRateEstimator(
         val delaySinceUpdate = (time - latestTime).toDouble / 1000
 
         // in elements/second
-        val processingRate = elements.toDouble / processingDelay * 1000
+        val processingRate = numElements.toDouble / processingDelay * 1000
 
+        // In our system `error` is the difference between the desired rate and the measured rate
+        // based on the latest batch information. We consider the desired rate to be latest rate,
+        // which is what this estimator calculated for the previous batch.
         // in elements/second
         val error = latestRate - processingRate
 
-        // The error integral, based on schedulingDelay as an indicator for accumulated errors
-        // a scheduling delay s corresponds to s * processingRate overflowing elements. Those
+        // The error integral, based on schedulingDelay as an indicator for accumulated errors.
+        // A scheduling delay s corresponds to s * processingRate overflowing elements. Those
         // are elements that couldn't be processed in previous batches, leading to this delay.
-        // We assume the processingRate didn't change too much.
-        // from the number of overflowing elements we can calculate the rate at which they would be
+        // In the following, we assume the processingRate didn't change too much.
+        // From the number of overflowing elements we can calculate the rate at which they would be
         // processed by dividing it by the batch interval. This rate is our "historical" error,
         // or integral part, since if we subtracted this rate from the previous "calculated rate",
         // there wouldn't have been any overflowing elements, and the scheduling delay would have
@@ -87,8 +102,8 @@ private[streaming] class PIDRateEstimator(
         // in elements/(second ^ 2)
         val dError = (error - latestError) / delaySinceUpdate
 
-        val newRate = (latestRate + proportional * error +
-                                    integral * historicalError +
+        val newRate = (latestRate - proportional * error -
+                                    integral * historicalError -
                                     derivative * dError).max(0.0)
         latestTime = time
         if (firstRun) {

streaming/src/main/scala/org/apache/spark/streaming/scheduler/rate/RateEstimator.scala

@@ -58,9 +58,9 @@ object RateEstimator {
   def create(conf: SparkConf, batchInterval: Duration): RateEstimator =
     conf.get("spark.streaming.backpressure.rateEstimator", "pid") match {
       case "pid" =>
-        val proportional = conf.getDouble("spark.streraming.backpressure.pid.proportional", -1.0)
-        val integral = conf.getDouble("spark.streraming.backpressure.pid.integral", -0.2)
-        val derived = conf.getDouble("spark.streraming.backpressure.pid.derived", 0.0)
+        val proportional = conf.getDouble("spark.streaming.backpressure.pid.proportional", 1.0)
+        val integral = conf.getDouble("spark.streaming.backpressure.pid.integral", 0.2)
+        val derived = conf.getDouble("spark.streaming.backpressure.pid.derived", 0.0)
         new PIDRateEstimator(batchInterval.milliseconds, proportional, integral, derived)
 
       case estimator =>

streaming/src/test/scala/org/apache/spark/streaming/scheduler/rate/PIDRateEstimatorSuite.scala

@@ -19,35 +19,64 @@ package org.apache.spark.streaming.scheduler.rate
 
 import scala.util.Random
 
-import org.scalatest._
+import org.scalatest.Inspectors.forAll
 import org.scalatest.Matchers
-import org.scalatest.Inspectors._
 
-import org.apache.spark.SparkFunSuite
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.streaming.Seconds
 
 class PIDRateEstimatorSuite extends SparkFunSuite with Matchers {
 
+  test("the right estimator is created") {
+    val conf = new SparkConf
+    conf.set("spark.streaming.backpressure.rateEstimator", "pid")
+    val pid = RateEstimator.create(conf, Seconds(1))
+    pid.getClass should equal(classOf[PIDRateEstimator])
+  }
+
+  test("estimator checks ranges") {
+    intercept[IllegalArgumentException] {
+      new PIDRateEstimator(0, 1, 2, 3)
+    }
+    intercept[IllegalArgumentException] {
+      new PIDRateEstimator(100, -1, 2, 3)
+    }
+    intercept[IllegalArgumentException] {
+      new PIDRateEstimator(100, 0, -1, 3)
+    }
+    intercept[IllegalArgumentException] {
+      new PIDRateEstimator(100, 0, 0, -1)
+    }
+  }
+
+  private def createDefaultEstimator: PIDRateEstimator = {
+    new PIDRateEstimator(20, 1D, 0D, 0D)
+  }
+
   test("first bound is None") {
-    val p = new PIDRateEstimator(20, -1D, 0D, 0D)
+    val p = createDefaultEstimator
     p.compute(0, 10, 10, 0) should equal(None)
   }
 
   test("second bound is rate") {
-    val p = new PIDRateEstimator(20, -1D, 0D, 0D)
+    val p = createDefaultEstimator
     p.compute(0, 10, 10, 0)
     // 1000 elements / s
     p.compute(10, 10, 10, 0) should equal(Some(1000))
   }
 
   test("works even with no time between updates") {
-    val p = new PIDRateEstimator(20, -1D, 0D, 0D)
+    val p = createDefaultEstimator
     p.compute(0, 10, 10, 0)
     p.compute(10, 10, 10, 0)
     p.compute(10, 10, 10, 0) should equal(None)
   }
 
   test("bound is never negative") {
-    val p = new PIDRateEstimator(20, -1D, -1D, 0D)
+    val p = new PIDRateEstimator(20, 1D, 1D, 0D)
+    // prepare a series of batch updates, one every 20ms, 0 processed elements, 2ms of processing
+    // this might point the estimator to try and decrease the bound, but we test it never
+    // goes below zero, which would be nonsensical.
     val times = List.tabulate(50)(x => x * 20) // every 20ms
     val elements = List.fill(50)(0) // no processing
     val proc = List.fill(50)(20) // 20ms of processing
@@ -58,7 +87,10 @@ class PIDRateEstimatorSuite extends SparkFunSuite with Matchers {
   }
 
   test("with no accumulated or positive error, |I| > 0, follow the processing speed") {
-    val p = new PIDRateEstimator(20, -1D, -1D, 0D)
+    val p = new PIDRateEstimator(20, 1D, 1D, 0D)
+    // prepare a series of batch updates, one every 20ms with an increasing number of processed
+    // elements in each batch, but constant processing time, and no accumulated error. Even though
+    // the integral part is non-zero, the estimated rate should follow only the proportional term
     val times = List.tabulate(50)(x => x * 20) // every 20ms
     val elements = List.tabulate(50)(x => x * 20) // increasing
     val proc = List.fill(50)(20) // 20ms of processing
@@ -69,7 +101,11 @@ class PIDRateEstimatorSuite extends SparkFunSuite with Matchers {
   }
 
   test("with no accumulated but some positive error, |I| > 0, follow the processing speed") {
-    val p = new PIDRateEstimator(20, -1D, -1D, 0D)
+    val p = new PIDRateEstimator(20, 1D, 1D, 0D)
+    // prepare a series of batch updates, one every 20ms with an decreasing number of processed
+    // elements in each batch, but constant processing time, and no accumulated error. Even though
+    // the integral part is non-zero, the estimated rate should follow only the proportional term,
+    // asking for less and less elements
     val times = List.tabulate(50)(x => x * 20) // every 20ms
     val elements = List.tabulate(50)(x => (50 - x) * 20) // decreasing
     val proc = List.fill(50)(20) // 20ms of processing
@@ -80,7 +116,7 @@ class PIDRateEstimatorSuite extends SparkFunSuite with Matchers {
   }
 
   test("with some accumulated and some positive error, |I| > 0, stay below the processing speed") {
-    val p = new PIDRateEstimator(20, -1D, -.01D, 0D)
+    val p = new PIDRateEstimator(20, 1D, .01D, 0D)
     val times = List.tabulate(50)(x => x * 20) // every 20ms
     val rng = new Random()
     val elements = List.tabulate(50)(x => rng.nextInt(1000))