[SPARK-8979][Streaming] Implements a PIDRateEstimator #17

huitseeker · 2015-07-14T14:49:01Z

This depends on #16
Derivation in https://www.dropbox.com/s/dwgl7wa1z5wbkg6/PIDderivation.pdf?dl=0

dragos · 2015-07-15T09:40:17Z

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

+   */
+  protected [streaming] val rateEstimator = ssc.conf
+                                               .getOption("spark.streaming.RateEstimator")
+                                               .getOrElse("noop") match {


I'd break this logic into its own method. The list of implementations might grow.

What do you suggest the signature of that method would be ?

How about String => RateEstimator?

tdas · 2015-07-15T20:41:11Z

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

+ * an estimate of the speed at which this stream should ingest messages,
+ * given an estimate computation from a `RateEstimator`
+ */
+@DeveloperApi


This must be private[streaming]

I assume the same goes for RateEstimator, NoopRateEstimator, and PIDRateEstimator, at least for now, right ?

tdas · 2015-07-15T21:44:52Z

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

+  @transient private val executionContext = ExecutionContext.fromExecutorService(
+    ThreadUtils.newDaemonSingleThreadExecutor("stream-rate-update"))
+
+  private val speedLimit : AtomicLong = new AtomicLong(-1L)


speedLimit --> rateLimit

…ements the ReceiverRateController

…into its own method

tdas · 2015-07-15T22:16:06Z

Here should be the test plan.

RateLimiterSuite - add unit test to test rate limit updates (mentioned in this in the inline comments)
ReceiverTrackerSuite - add unit test that tests the communication between receiver - (i) runs a streamingContext+receiverStream+fake receiver that when started gives us access to the internal rate limiter of the receiver (may have to use objects) (ii) directly change the rate by calling receiverTracker.updateRate (iii) check the current rate in the receiver's rate limiter (not by sending data, but directly)

These should be part of subtask 1 or dynamic rate limiting. Additionally, with the rate controller subtask.

RateControllerSuite -

a. add unit test that uses a mock RateController to test whether publish() gets called when onBatchCompleted is called

b. add unit test that uses a fake RateEstimator (something that just doubles the rate) to test whether the rate limit is being set correctly in the Receiver. Use the technique explained above in 2.
PIDRateEstimatorSuite - already there, may need changes but i will review it later.

tdas · 2015-07-15T22:42:54Z

Overall, this whole PR looks quite good and ready for PRs to the Spark main repo. I suggest breaking them down into smaller PRs according to the subtasks.
@huitseeker @dragos
Thank you very much for your efforts. :)

typesafe-tools · 2015-07-16T05:42:37Z

Refer to this link for build results (access rights to CI server needed):
https://ci.typesafe.com/job/ghprb-spark-multi-conf/52/

Build Log
last 10 lines

[...truncated 11 lines...]
 > git rev-parse refs/remotes/origin/origin/pr/17/merge^{commit} # timeout=10
Checking out Revision 16437f83d698c6f2d2a7572a87a4ba0a91e72636 (refs/remotes/origin/pr/17/merge)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 16437f83d698c6f2d2a7572a87a4ba0a91e72636
First time build. Skipping changelog.
Triggering ghprb-spark-multi-conf » Spark-Ora-JDK7-PV,2.10
ghprb-spark-multi-conf » Spark-Ora-JDK7-PV,2.10 completed with result FAILURE
Notifying upstream projects of job completion
Setting status of 48c869c75acf5c613e3e737b3913a189af0dada5 to FAILURE with url http://ci.typesafe.com/job/ghprb-spark-multi-conf/52/ and message: Merged build finished.

Test FAILed.

typesafe-tools · 2015-07-16T10:06:38Z

Refer to this link for build results (access rights to CI server needed):
https://ci.typesafe.com/job/ghprb-spark-multi-conf/54/

Build Log
last 10 lines

[...truncated 11 lines...]
 > git rev-parse refs/remotes/origin/origin/pr/17/merge^{commit} # timeout=10
Checking out Revision 16437f83d698c6f2d2a7572a87a4ba0a91e72636 (refs/remotes/origin/pr/17/merge)
 > git config core.sparsecheckout # timeout=10
 > git checkout -f 16437f83d698c6f2d2a7572a87a4ba0a91e72636
First time build. Skipping changelog.
Triggering ghprb-spark-multi-conf » Spark-Ora-JDK7-PV,2.10
ghprb-spark-multi-conf » Spark-Ora-JDK7-PV,2.10 completed with result FAILURE
Notifying upstream projects of job completion
Setting status of 94c2225bb60e800194cdacde35b6916678bc676d to FAILURE with url http://ci.typesafe.com/job/ghprb-spark-multi-conf/54/ and message: Merged build finished.

Test FAILed.

…onfig option. ## What changes were proposed in this pull request? Currently, `OptimizeIn` optimizer replaces `In` expression into `InSet` expression if the size of set is greater than a constant, 10. This issue aims to make a configuration `spark.sql.optimizer.inSetConversionThreshold` for that. After this PR, `OptimizerIn` is configurable. ```scala scala> sql("select a in (1,2,3) from (select explode(array(1,2)) a) T").explain() == Physical Plan == WholeStageCodegen : +- Project [a#7 IN (1,2,3) AS (a IN (1, 2, 3))#8] : +- INPUT +- Generate explode([1,2]), false, false, [a#7] +- Scan OneRowRelation[] scala> sqlContext.setConf("spark.sql.optimizer.inSetConversionThreshold", "2") scala> sql("select a in (1,2,3) from (select explode(array(1,2)) a) T").explain() == Physical Plan == WholeStageCodegen : +- Project [a#16 INSET (1,2,3) AS (a IN (1, 2, 3))#17] : +- INPUT +- Generate explode([1,2]), false, false, [a#16] +- Scan OneRowRelation[] ``` ## How was this patch tested? Pass the Jenkins tests (with a new testcase) Author: Dongjoon Hyun <dongjoon@apache.org> Closes apache#12562 from dongjoon-hyun/SPARK-14796.

huitseeker force-pushed the SPARK-8979 branch 3 times, most recently from 94c2225 to 2144472 Compare July 15, 2015 00:34

huitseeker changed the title ~~[SPARK-8979][Streaming] Implement a PIDRateEstimator~~ [SPARK-8979][Streaming] Implements a PIDRateEstimator Jul 15, 2015

dragos reviewed Jul 15, 2015
View reviewed changes

huitseeker force-pushed the SPARK-8979 branch 6 times, most recently from 219e1bd to ef7b0a4 Compare July 15, 2015 18:51

tdas reviewed Jul 15, 2015
View reviewed changes

huitseeker force-pushed the SPARK-8979 branch from ef7b0a4 to 9076fc6 Compare July 15, 2015 21:39

tdas reviewed Jul 15, 2015
View reviewed changes

huitseeker added 3 commits July 15, 2015 23:57

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

42636ee

…ements the ReceiverRateController

[SPARK-8979][Streaming] Implements a PIDRateEstimator

d8c41cb

[SPARK-8979][Streaming] Break the logic to instantiate rateEstimator …

f5c1492

…into its own method

huitseeker force-pushed the SPARK-8979 branch from 9076fc6 to f5c1492 Compare July 15, 2015 21:59

dragos force-pushed the topic/streaming-bp/dynamic-rate branch from 13ada97 to 0c51959 Compare July 20, 2015 14:45

dragos closed this Sep 9, 2015

[SPARK-8979][Streaming] Implements a PIDRateEstimator #17

[SPARK-8979][Streaming] Implements a PIDRateEstimator #17

huitseeker commented Jul 14, 2015

dragos Jul 15, 2015

huitseeker Jul 15, 2015

dragos Jul 15, 2015

tdas Jul 15, 2015

huitseeker Jul 15, 2015

tdas Jul 15, 2015

huitseeker Jul 15, 2015

tdas Jul 15, 2015

tdas commented Jul 15, 2015

tdas commented Jul 15, 2015

typesafe-tools commented Jul 16, 2015

typesafe-tools commented Jul 16, 2015

[SPARK-8979][Streaming] Implements a PIDRateEstimator #17

[SPARK-8979][Streaming] Implements a PIDRateEstimator #17

Conversation

huitseeker commented Jul 14, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tdas commented Jul 15, 2015

tdas commented Jul 15, 2015

typesafe-tools commented Jul 16, 2015

typesafe-tools commented Jul 16, 2015