[SPARK-8977][Streaming] Defines the RateEstimator interface, and impements the RateController #7600

dragos · 2015-07-22T21:27:19Z

Based on #7471.

add a test that exercises the publish path from driver to receiver
remove Serializable from RateController and RateEstimator

…river to the block generator

…aming.receiver.maxRate

[SPARK-8975][Streaming] Adds a mechanism to send a new rate from the driver to the block generator

…eivers.

- made rate limit a Long and default to Long.MaxValue (consequence of the above) - removed custom `waitUntil` and replaced it by `eventually`

As I mentioned before, I don’t think this is a great idea: - such tests are flaky (original test in ReceiverSuite was ignored for that reason) - Guava’s code has its own test suite, so we can assume it implements `setRate` correctly I noticed one flaky failure in about 10 runs on my machine (receiver got 1 message less than the lower bound, which is within 5% of the nominal rate).

This reverts commit 0c51959.

use the listener bus to know when receivers have registered (`onStart` is called before receivers have registered, leading to flaky behavior).

…ements the ReceiverRateController

dragos · 2015-07-22T21:29:30Z

streaming/src/main/scala/org/apache/spark/streaming/scheduler/RateController.scala

+ */
+@DeveloperApi
+private [streaming] abstract class RateController(val streamUID: Int, rateEstimator: RateEstimator)
+  extends StreamingListener with Serializable {


Probably not worth to serialize (same for RateEstimator). If we make them @transient, we should add a test that after recovery from a checkpoint the rate controller is still active (non-null and registered with the streaming listener bus).

@tdas what do you think?

Yes. Agreed. Transient plus update some existing test case in CheckpointSuite

SparkQA · 2015-07-22T21:36:30Z

Test build #38112 has finished for PR 7600 at commit 8b08492.

This patch fails RAT tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-07-22T22:35:14Z

Test build #38115 has finished for PR 7600 at commit 34a389d.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

tdas · 2015-07-23T04:16:39Z

streaming/src/main/scala/org/apache/spark/streaming/dstream/InputDStream.scala

+   * A rate estimator configured by the user to compute a dynamic ingestion bound for this stream.
+   * @see `RateEstimator`
+   */
+  protected [streaming] val rateEstimator = newEstimator()


Why do you need rateEstimator as a field in DStream? All you need expose to modules outside DStream is the RateController. And the RateEstimator can be within the RateController

…p/rate-controller

SparkQA · 2015-07-23T11:06:32Z

Test build #38215 has finished for PR 7600 at commit 238cfc6.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

estimator, changed logic for initialising rate estimator.

SparkQA · 2015-07-23T15:20:14Z

Test build #38228 has finished for PR 7600 at commit b425d32.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

with several rate updates.

dragos · 2015-07-23T16:52:05Z

These tests took a while to debug, so I'll remove the serializable components tonight or tomorrow (it's a pretty localized and isolated change). In the meantime please have a look at the current state.

SparkQA · 2015-07-23T17:19:40Z

Test build #38243 has finished for PR 7600 at commit e57c66b.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

tdas · 2015-07-24T06:46:50Z

streaming/src/main/scala/org/apache/spark/streaming/dstream/InputDStream.scala

@@ -47,6 +49,14 @@ abstract class InputDStream[T: ClassTag] (@transient ssc_ : StreamingContext)
  /** This is an unique identifier for the input stream. */
  val id = ssc.getNewInputStreamId()

+  // Keep track of the freshest rate for this stream using the rateEstimator
+  protected[streaming] val rateController: Option[RateController] =
+    RateEstimator.makeEstimator(ssc.conf).map { estimator =>


This should change a little. See comment on what the config params should be in another comment below (near RateEstimator.makeEstimator)

And InputDStream should not have a RateController by default. So this should be None. Its will be upto implementation of InputDStream to override and define.

…p/rate-controller

SparkQA · 2015-07-28T10:54:07Z

Test build #38696 has finished for PR 7600 at commit 83357bd.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- public static class StructWriter
- case class CreateStructUnsafe(children: Seq[Expression]) extends Expression
- case class CreateNamedStructUnsafe(children: Seq[Expression]) extends Expression
- case class TungstenProject(projectList: Seq[NamedExpression], child: SparkPlan) extends UnaryNode

tdas · 2015-07-28T13:06:27Z

streaming/src/main/scala/org/apache/spark/streaming/dstream/ReceiverInputDStream.scala

@@ -41,6 +42,16 @@ abstract class ReceiverInputDStream[T: ClassTag](@transient ssc_ : StreamingCont
  extends InputDStream[T](ssc_) {

  /**
+   * Asynchronously maintains & sends new rate limits to the receiver through the receiver tracker.
+   */
+  override protected[streaming] val rateController: Option[RateController] =


nit: Please use { and } for multiline methods.

dragos · 2015-07-28T16:41:42Z

@tdas I have some comments in my email that don't appear here, so I suppose you changed your mind and deleted it.

I think the review would be smoother if you explained a bit better what are your underlying concerns, rather than very specific points. For instance, I don't understand why you insist I should avoid using your own helper methods. I don't think code would get clearer, nor less flaky. If anything, reusing these methods makes the code more uniform and easier to understand, plus any bugs that get fixed in those helpers benefit this test suite too.

SparkQA · 2015-07-28T17:15:44Z

Test build #38719 has finished for PR 7600 at commit 5125e60.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

tdas · 2015-07-28T19:41:36Z

@dragos I didnt delete any comment! The comments may have been to earlier threads which are currently hidden due to outdated diffs.

I know that these TestSuiteBase are the helper methods that I have provided with Spark Streaming. They were done a long time ago and since then we have learnt quite a bit about the shortcoming of those methods (runStreams), and when to use them and when its cleaner to not use them. The problem with runStreams is that

it hides part of the management of life cycle of the context. It starts and stop the StreamingContext, which give little control when you want to optimize stuff like.

Share SparkContexts across StreamingContexts is not possible through this as the Improves test times if we dont have to spin up SparkContext for every unit test.
Tests have unnecessary delays when designed using runStreams. This you see for yourself. With or without runStreams you have to use eventually to test your condition. So why wait for the runStreams to complete (especially when running a real clock) and then test with eventually? And how do you decide how many batches to run in runStreams? Is it even important to care about the number of batches to run considering the test really wants to verify some other condition being eventually true? Also why go to the complexity of attaching a TestOutputDStream if the test does care about the output? It will improve test times as well simplify the test if in these cases you just test with eventually alone, without using abstractions that does not really help.

Look at all the newly updated streaming testsuitesl, none of them use runStreams. For a reason. Hope this clarifies why I am trying to move the code away from runStreams.

dragos · 2015-07-28T20:17:59Z

On 28 iul. 2015, at 21:42, Tathagata Das notifications@github.com wrote:

@dragos I didnt delete any comment! The comments may have been to earlier threads which are currently hidden due to outdated diffs.

Weird. Probably you attached it to a commit that went away. They seemed to refer to code that was no longer there, but still new.
I know that these TestSuiteBase are the helper methods that I have provided with Spark Streaming. They were done a long time ago and since then we have learnt quite a bit about the shortcoming of those methods (runStreams), and when to use them and when its cleaner to not use them.

How about you deprecate them? It would save time for everyone touching spark streaming.
Tests have unnecessary delays when designed using runStreams. This you see for yourself. With or without runStreams you have to use eventually to test your condition. So why wait for the runStreams to complete (especially when running a real clock) and then test with eventually?
If you run on a real clock, I don't see what you gain in terms of time. It's not faster, is it?
And how do you decide how many batches to run in runStreams?
It's quite simple: in my test I have 3 rates I want to observe, so I need at least 3 batches.
Is it even important to care about the number of batches to run considering the test really wants to verify some other condition being eventually true?
Yes, since I need to make sure I observe the right number of rate updates, with the right values.
Also why go to the complexity of attaching a TestOutputDStream if the test does care about the output?
The 'complexity' we're talking about is 2 lines of code. I'm happy to be proven wrong, but inlining the useful parts of runStreams will be more than that.

It will improve test times as well simplify the test if in these cases you just test with eventually alone, without using abstractions that does not really help.
I think they help, but again, happy to be proven wrong.

How about you rewrite the last test and compare? If you're still convinced it's better I'll rewrite the rest.

Look at all the newly updated streaming testsuitesl, none of them use runStreams. For a reason. Hope this clarifies why I am trying to move the code away from runStreams.

—
Reply to this email directly or view it on GitHub.

tdas · 2015-07-28T20:31:42Z

streaming/src/test/scala/org/apache/spark/streaming/scheduler/RateControllerSuite.scala

+          } catch {
+            case NonFatal(_) => () // don't stop if the executor wasn't installed yet
+          }
+          Thread.sleep(20)


This is going to be a flaky test. There is no guarantee that this 20 ms sleep will be 20 ms. Can very well be 100 ms. And it cannot be guaranteed that will always get a the 100, 200, 300 will be received by receiver by the time 4 batches are over through runStreams. This going to be SUPER FLAKY. The non-flaky way of doing this

run streamingContext normally in real time (no real need for runStreams, just ssc.start())

test using eventually { observedRates should contain theSameElementsAs (rates :+ Long.MaxValue) }, no need to run anything in the background and all.

observedRates must be a synchronized hashset.

Ok, in the interest of closing this PR I will do the changes you request ASAP.

However, in the interest of understanding, I ran it 100 times without it failing once. If you have evidence of it being SUPER FLAKY in practice, do you mind sharing it here?

I guess the reason I'm not seeing these failures is that even if theoretically the delay could be higher, say 100ms, since the batch interval is 1s, you'd still sample it 10 times. And looking at the implementation of SystemClock, it's still based on Thread.sleep, so any systematic delay (say, system under heavy load) will affect both this test and the batch interval sampling rate.

BTW, eventually is using Thread.sleep as well (15ms between attempts, by default), so I guess the same considerations regarding flakiness should apply.

tdas · 2015-07-28T20:40:01Z

Cannot deprecate TestSuiteBase as it is still perfectly suited for testsuites like BasicOperationsSuite where the exact per-batch output needs to be verified. Its suitable for some use cases, just not all usecase, especially not for those where the output is not important.
I visited the new tests in RateControllerSuite again
(a)"rate controller publishes updates": No need for runStreams on exact number of batches, all you need is eventually { publishCalls > 0 }
(b) "publish rates reach receivers": No need for runStreams on exact number of batches, all you need is eventually {dstream.getCurrentRateLimit === Some(200) }
(c) You are trying to run 4 batches and expect there will 3 observed rates immediately as the 4 batches complete. This CANNOT be guaranteed. Thread.sleep cannot guarantee that you observe the rate limits at the right times so that you observe all the different rates. See comments there, I have suggest how to make this non-flaky. Again, does not require runStreams with exact number of batches.

tdas · 2015-07-28T20:53:57Z

streaming/src/test/scala/org/apache/spark/streaming/scheduler/ReceiverTrackerSuite.scala


 /**
 * Dummy receiver implementation
 */
-private class DummyReceiver(receiverId: Int, host: Option[String] = None)
+private[streaming] class DummyReceiver(receiverId: Int, host: Option[String] = None)


Since this receiver and associated InputDStream is used in multiple testsuites, mind renaming this to a more contextually relevant name than Dummy? How about RateTestReceiver and RateTestInputDStream?

Also you can augment these classes to count the number of times rates have been received, so that the MockRateLimitDStream can be removed. Trying to minimize the number of new classes we create.

SparkQA · 2015-07-29T10:50:47Z

Test build #38833 has finished for PR 7600 at commit f168c94.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

dragos · 2015-07-29T13:03:01Z

A summary of my changes and a few notes, in the hope of speeding up the process:

tests are bit faster after moving away from runStreams, so I stand corrected
Tests are still extending TestSuiteBase because withStreamingContext is defined in there
I moved the sampling of observedRates inside the eventually loop, as I don't know how to do it otherwise without using another thread
I didn't make the HashSet synchronized, since there's no background thread anymore

The rest of the changes should match your requests and I don't list them here.

SparkQA · 2015-07-29T19:16:06Z

Test build #1226 has finished for PR 7600 at commit f168c94.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2015-07-29T19:16:24Z

Test build #1225 has finished for PR 7600 at commit f168c94.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class RequestExecutors(
- class RFormula(override val uid: String) extends Estimator[RFormulaModel] with RFormulaBase
- public static class StructWriter
- abstract class InternalRow extends Serializable with SpecializedGetters
- case class CreateStructUnsafe(children: Seq[Expression]) extends Expression
- case class CreateNamedStructUnsafe(children: Seq[Expression]) extends Expression
- case class LastDay(startDate: Expression) extends UnaryExpression with ImplicitCastInputTypes
- case class NextDay(startDate: Expression, dayOfWeek: Expression)
- case class TungstenProject(projectList: Seq[NamedExpression], child: SparkPlan) extends UnaryNode

tdas · 2015-07-29T20:46:43Z

LGTM There are still an issues that occurred to me in the last moment, but I am inclined to merge this as is to unblock other PRs - RateController should not be serialized with checkpoints. Does not make sense to recover the RateController. Can you make a PR for this?

Regarding flakiness, it often times occurs only in Jenkins as things can run very slowly in Jenkins, and causes all this Thread.sleep related flakiness to occur. But your current method without background thread make it less flaky, but still does not guarantee. I thought about the test. The least flaky way to test that multiple rate update reach the receiver, would be to send one rate update, wait for it to be received, then send another update with different rate, then wait for it.

tdas · 2015-07-29T20:48:15Z

Aright I merged this. Thanks all!

dragos · 2015-07-30T07:47:24Z

LGTM There are still an issues that occurred to me in the last moment, but I am inclined to merge this as is to unblock other PRs - RateController should not be serialized with checkpoints. Does not make sense to recover the RateController. Can you make a PR for this?

I forgot to mention this. I spent some time on it, but it looked like the gains were not worth it. The problem is deserialization, and the specific RateEstimator instance. At deserialization time there's no ssc available yet (it's null), so there's no easy way to get to a SparkConf in order to get the configured estimator. We could work around that by attaching the controller at a later point, during start or some other lifecycle event, but it seemed to complicate matters quite a bit, for little gain that I can see. Thanks for not blocking the PR over this!

Based on #7600 /cc tdas Author: Iulian Dragos <jaguarul@gmail.com> Author: François Garillot <francois@garillot.net> Closes #7648 from dragos/topic/streaming-bp/pid and squashes the following commits: aa5b097 [Iulian Dragos] Add more comments, made all PID constant parameters positive, a couple more tests. 93b74f8 [Iulian Dragos] Better explanation of historicalError. 7975b0c [Iulian Dragos] Add configuration for PID. 26cfd78 [Iulian Dragos] A couple of variable renames. d0bdf7c [Iulian Dragos] Update to latest version of the code, various style and name improvements. d58b845 [François Garillot] [SPARK-8979][Streaming] Implements a PIDRateEstimator

huitseeker and others added 10 commits July 15, 2015 23:17

[SPARK-8975][Streaming] Add a mechanism to send a new rate from the d…

4721c7d

…river to the block generator

[SPARK-8975][Streaming] Adds Ratelimiter unit tests w.r.t. spark.stre…

d15de42

…aming.receiver.maxRate

Merge pull request #15 from huitseeker/SPARK-8975

6369b30

[SPARK-8975][Streaming] Adds a mechanism to send a new rate from the driver to the block generator

Add a test for the propagation of a new rate limit from driver to rec…

cd1397d

…eivers.

- removed field to hold the current rate limit in rate limiter

261a051

- made rate limit a Long and default to Long.MaxValue (consequence of the above) - removed custom `waitUntil` and replaced it by `eventually`

Revert "Added a few tests that measure the receiver’s rate."

210f495

This reverts commit 0c51959.

Use Reflection for accessing truly private executor method and

162d9e5

use the listener bus to know when receivers have registered (`onStart` is called before receivers have registered, leading to flaky behavior).

Renames and other nitpicks.

8941cf9

[SPARK-8977][Streaming] Defines the RateEstimator interface, and impl…

d32ca36

…ements the ReceiverRateController

dragos reviewed Jul 22, 2015
View reviewed changes

Various style changes and a first test for the rate controller.

34a389d

dragos force-pushed the topic/streaming-bp/rate-controller branch from 8b08492 to 34a389d Compare July 22, 2015 21:43

tdas reviewed Jul 23, 2015
View reviewed changes

Merge remote-tracking branch 'upstream/master' into topic/streaming-b…

238cfc6

…p/rate-controller

Removed DeveloperAPI, removed rateEstimator field, removed Noop rate

b425d32

estimator, changed logic for initialising rate estimator.

Added a couple of tests for the full scenario from driver to receivers,

e57c66b

with several rate updates.

tdas reviewed Jul 24, 2015
View reviewed changes

dragos added 2 commits July 28, 2015 11:47

Latest round of reviews.

475e346

Merge remote-tracking branch 'upstream/master' into topic/streaming-b…

a2eb3b9

…p/rate-controller

tdas reviewed Jul 28, 2015
View reviewed changes

Fix style.

5125e60

dragos force-pushed the topic/streaming-bp/rate-controller branch from 83357bd to 5125e60 Compare July 28, 2015 16:38

tdas reviewed Jul 28, 2015
View reviewed changes

Latest review round.

f168c94

asfgit closed this in 819be46 Jul 29, 2015

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-8977][Streaming] Defines the RateEstimator interface, and impements the RateController #7600

[SPARK-8977][Streaming] Defines the RateEstimator interface, and impements the RateController #7600

dragos commented Jul 22, 2015

dragos Jul 22, 2015

tdas Jul 23, 2015

SparkQA commented Jul 22, 2015

SparkQA commented Jul 22, 2015

tdas Jul 23, 2015

SparkQA commented Jul 23, 2015

SparkQA commented Jul 23, 2015

dragos commented Jul 23, 2015

SparkQA commented Jul 23, 2015

tdas Jul 24, 2015

tdas Jul 24, 2015

SparkQA commented Jul 28, 2015

tdas Jul 28, 2015

dragos commented Jul 28, 2015

SparkQA commented Jul 28, 2015

tdas commented Jul 28, 2015

dragos commented Jul 28, 2015

tdas Jul 28, 2015

dragos Jul 29, 2015

tdas commented Jul 28, 2015

tdas Jul 28, 2015

tdas Jul 28, 2015

SparkQA commented Jul 29, 2015

dragos commented Jul 29, 2015

SparkQA commented Jul 29, 2015

SparkQA commented Jul 29, 2015

tdas commented Jul 29, 2015

tdas commented Jul 29, 2015

dragos commented Jul 30, 2015

[SPARK-8977][Streaming] Defines the RateEstimator interface, and impements the RateController #7600

[SPARK-8977][Streaming] Defines the RateEstimator interface, and impements the RateController #7600

Conversation

dragos commented Jul 22, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jul 22, 2015

SparkQA commented Jul 22, 2015

Choose a reason for hiding this comment

SparkQA commented Jul 23, 2015

SparkQA commented Jul 23, 2015

dragos commented Jul 23, 2015

SparkQA commented Jul 23, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jul 28, 2015

Choose a reason for hiding this comment

dragos commented Jul 28, 2015

SparkQA commented Jul 28, 2015

tdas commented Jul 28, 2015

dragos commented Jul 28, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tdas commented Jul 28, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jul 29, 2015

dragos commented Jul 29, 2015

SparkQA commented Jul 29, 2015

SparkQA commented Jul 29, 2015

tdas commented Jul 29, 2015

tdas commented Jul 29, 2015

dragos commented Jul 30, 2015