You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by lw-lin <gi...@git.apache.org> on 2016/04/30 04:03:13 UTC
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
GitHub user lw-lin opened a pull request:
https://github.com/apache/spark/pull/12797
[SPARK-15022][SPARK-15023] Add support for testing against the `ProcessingTime(intervalMS > 0)` trigger and `ManualClock`
## What changes were proposed in this pull request?
Currently in `StreamTest`, we have a `StartStream` which will start a streaming query against trigger `ProcessTime(intervalMS = 0)` and `SystemClock`.
We also need to test cases against `ProcessTime(intervalMS > 0)`, which often requires `ManualClock`.
This patch:
- fixes an issue of `ProcessingTimeExecutor`, where for a batch it should run `batchRunner` only once but might run multiple times under certain conditions;
- adds support for testing against the `ProcessingTime(intervalMS > 0)` trigger and `ManualClock`, by specifying them as fields for `StartStream`, and by adding an `AdvanceClock` action;
- adds a test, which takes advantage of the new `StartStream` and `AdvanceClock`, to test against [PR#[SPARK-14942] Reduce delay between batch construction and execution ](https://github.com/apache/spark/pull/12725).
## How was this patch tested?
N/A
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/lw-lin/spark add-trigger-test-support
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/12797.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #12797
----
commit 90ed69285fc34ae43a0f454ceb25837618212e28
Author: Liwei Lin <lw...@gmail.com>
Date: 2016-04-30T01:32:48Z
fix an issue of nextBatchTime against ManualClock
commit 9d80b15e33151f5f87207d72f89f016c18a21b01
Author: Liwei Lin <lw...@gmail.com>
Date: 2016-04-30T01:33:48Z
Add support for testing against ProcessingTime(intervalMS>0)
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61771197
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala ---
@@ -65,8 +65,22 @@ case class ProcessingTimeExecutor(processingTime: ProcessingTime, clock: Clock =
s"${intervalMs} milliseconds, but spent ${realElapsedTimeMs} milliseconds")
}
- /** Return the next multiple of intervalMs */
+ /** Return the next multiple of intervalMs
--- End diff --
Nit: comment style is off, we use [javadoc style](https://cwiki.apache.org/confluence/display/SPARK/Spark+Code+Style+Guide#SparkCodeStyleGuide-Codedocumentationstyle)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61774572
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ProcessingTimeExecutorSuite.scala ---
@@ -21,19 +21,41 @@ import java.util.concurrent.{CountDownLatch, TimeUnit}
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.ProcessingTime
-import org.apache.spark.util.ManualClock
+import org.apache.spark.util.{Clock, ManualClock, SystemClock}
class ProcessingTimeExecutorSuite extends SparkFunSuite {
test("nextBatchTime") {
val processingTimeExecutor = ProcessingTimeExecutor(ProcessingTime(100))
+ assert(processingTimeExecutor.nextBatchTime(0) === 100)
assert(processingTimeExecutor.nextBatchTime(1) === 100)
assert(processingTimeExecutor.nextBatchTime(99) === 100)
- assert(processingTimeExecutor.nextBatchTime(100) === 100)
+ assert(processingTimeExecutor.nextBatchTime(100) === 200)
assert(processingTimeExecutor.nextBatchTime(101) === 200)
assert(processingTimeExecutor.nextBatchTime(150) === 200)
}
+ private def testNextBatchTimeAgainstClock(clock: Clock) {
+ val IntervalMS = 100
--- End diff --
lowercase first letter for variables.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-216309861
Some minor comments about code understandability, but overall this looks good. Thanks for working on this!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-216616857
Looks pretty good. @lw-lin could you address the comments and resolve the conflicts?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61774969
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ProcessingTimeExecutorSuite.scala ---
@@ -21,19 +21,41 @@ import java.util.concurrent.{CountDownLatch, TimeUnit}
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.ProcessingTime
-import org.apache.spark.util.ManualClock
+import org.apache.spark.util.{Clock, ManualClock, SystemClock}
class ProcessingTimeExecutorSuite extends SparkFunSuite {
test("nextBatchTime") {
val processingTimeExecutor = ProcessingTimeExecutor(ProcessingTime(100))
+ assert(processingTimeExecutor.nextBatchTime(0) === 100)
assert(processingTimeExecutor.nextBatchTime(1) === 100)
assert(processingTimeExecutor.nextBatchTime(99) === 100)
- assert(processingTimeExecutor.nextBatchTime(100) === 100)
+ assert(processingTimeExecutor.nextBatchTime(100) === 200)
assert(processingTimeExecutor.nextBatchTime(101) === 200)
assert(processingTimeExecutor.nextBatchTime(150) === 200)
}
+ private def testNextBatchTimeAgainstClock(clock: Clock) {
+ val IntervalMS = 100
+ val processingTimeExecutor = ProcessingTimeExecutor(ProcessingTime(IntervalMS), clock)
+
+ val ITERATION = 10
+ var nextBatchTime: Long = 0
+ for (it <- 1 to ITERATION)
+ nextBatchTime = processingTimeExecutor.nextBatchTime(nextBatchTime)
+
+ // nextBatchTime should be 1000
+ assert(nextBatchTime === IntervalMS * ITERATION)
--- End diff --
What is this checking that isn't checked by `test("nextBatchTime")` above?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-216714511
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/57691/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61837727
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala ---
@@ -65,8 +65,22 @@ case class ProcessingTimeExecutor(processingTime: ProcessingTime, clock: Clock =
s"${intervalMs} milliseconds, but spent ${realElapsedTimeMs} milliseconds")
}
- /** Return the next multiple of intervalMs */
+ /** Return the next multiple of intervalMs
+ *
+ * e.g. for intervalMs = 100
+ * nextBatchTime(0) = 100
+ * nextBatchTime(1) = 100
+ * ...
+ * nextBatchTime(99) = 100
+ * nextBatchTime(100) = 200
+ * nextBatchTime(101) = 200
+ * ...
+ * nextBatchTime(199) = 200
+ * nextBatchTime(200) = 300
+ *
+ * Note, this way, we'll get nextBatchTime(nextBatchTime(0)) = 200, rather than = 0
--- End diff --
Let me update it with your much clearer verson! Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61976475
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/StreamTest.scala ---
@@ -142,7 +142,10 @@ trait StreamTest extends QueryTest with Timeouts {
case object StopStream extends StreamAction with StreamMustBeRunning
/** Starts the stream, resuming if data has already been processed. It must not be running. */
- case object StartStream extends StreamAction
+ case class StartStream(trigger: Trigger = null, triggerClock: Clock = null) extends StreamAction
--- End diff --
This layer of `null`s was intended to delegate the default values of `StreamExecution` into these tests, so that we don't have to set the same default values in many places and maintain their consistency. But since it seems very unlikely that we would change the default values, so I've removed the `null`s and followed your comments.
Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61774375
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala ---
@@ -65,8 +65,22 @@ case class ProcessingTimeExecutor(processingTime: ProcessingTime, clock: Clock =
s"${intervalMs} milliseconds, but spent ${realElapsedTimeMs} milliseconds")
}
- /** Return the next multiple of intervalMs */
+ /** Return the next multiple of intervalMs
+ *
+ * e.g. for intervalMs = 100
+ * nextBatchTime(0) = 100
+ * nextBatchTime(1) = 100
+ * ...
+ * nextBatchTime(99) = 100
+ * nextBatchTime(100) = 200
+ * nextBatchTime(101) = 200
+ * ...
+ * nextBatchTime(199) = 200
+ * nextBatchTime(200) = 300
+ *
+ * Note, this way, we'll get nextBatchTime(nextBatchTime(0)) = 200, rather than = 0
+ * */
def nextBatchTime(now: Long): Long = {
- (now - 1) / intervalMs * intervalMs + intervalMs
+ now / intervalMs * intervalMs + intervalMs
--- End diff --
When I wrote this method, I was trying to deal with one case: If a batch takes exactly `intervalMs`, we should run the next batch at once instead of sleeping `intervalMs`. This changes will break it.
However, I forgot to handle the case that a batch takes 0ms. How about changing [this line](https://github.com/apache/spark/blob/0513c3ac93e0a25d6eedbafe6c0561e71c92880a/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala#L53) to:
```Scala
if (batchElapsedTimeMs == 0) {
clock.waitTillTime(intervalMs)
} else {
clock.waitTillTime(nextBatchTime(batchEndTimeMs))
}
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-216705209
@marmbrus @zsxwing thank you for the patient review! :-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-216939189
Merging to master / 2.0. Thanks again @lw-lin
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61928352
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala ---
@@ -65,8 +65,22 @@ case class ProcessingTimeExecutor(processingTime: ProcessingTime, clock: Clock =
s"${intervalMs} milliseconds, but spent ${realElapsedTimeMs} milliseconds")
}
- /** Return the next multiple of intervalMs */
+ /** Return the next multiple of intervalMs
+ *
+ * e.g. for intervalMs = 100
+ * nextBatchTime(0) = 100
+ * nextBatchTime(1) = 100
+ * ...
+ * nextBatchTime(99) = 100
+ * nextBatchTime(100) = 200
+ * nextBatchTime(101) = 200
+ * ...
+ * nextBatchTime(199) = 200
+ * nextBatchTime(200) = 300
+ *
+ * Note, this way, we'll get nextBatchTime(nextBatchTime(0)) = 200, rather than = 0
+ * */
def nextBatchTime(now: Long): Long = {
- (now - 1) / intervalMs * intervalMs + intervalMs
+ now / intervalMs * intervalMs + intervalMs
--- End diff --
I see. I think your approach is better. Thanks for your clarifying.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61837469
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala ---
@@ -65,8 +65,22 @@ case class ProcessingTimeExecutor(processingTime: ProcessingTime, clock: Clock =
s"${intervalMs} milliseconds, but spent ${realElapsedTimeMs} milliseconds")
}
- /** Return the next multiple of intervalMs */
+ /** Return the next multiple of intervalMs
+ *
+ * e.g. for intervalMs = 100
+ * nextBatchTime(0) = 100
+ * nextBatchTime(1) = 100
+ * ...
+ * nextBatchTime(99) = 100
+ * nextBatchTime(100) = 200
+ * nextBatchTime(101) = 200
+ * ...
+ * nextBatchTime(199) = 200
+ * nextBatchTime(200) = 300
+ *
+ * Note, this way, we'll get nextBatchTime(nextBatchTime(0)) = 200, rather than = 0
+ * */
def nextBatchTime(now: Long): Long = {
- (now - 1) / intervalMs * intervalMs + intervalMs
+ now / intervalMs * intervalMs + intervalMs
--- End diff --
@zsxwing thanks for clarifying on this! :-)
[1]
The issue is triggered when both `batchElapsedTimeMs == 0` and `batchEndTimeMs` is multiple of `intervalMS` hold, e.g. `batchStartTimeMs == 50` and `batchEndTimeMS == 50` given `intervalMS == 100` won't trigger the issue. So, we might have to do like this:
```scala
if (batchElapsedTimeMs == 0 && batchEndTimeMs % intervalMS == 0) {
clock.waitTillTime(batchEndTimeMs + intervalMs)
} else {
clock.waitTillTime(nextBatchTime(batchEndTimeMs))
}
```
For me It seems a little hard to interpret...
[2]
>
... deal with one case: If a batch takes exactly intervalMs, we should run the next batch at once instead of sleeping intervalMs
This is a good point! I've done some calculations based on your comments, and it seems we would still run the next batch at once when the last job takes exactly `intervalMs`?
prior to this path:
```
batch | job
-----------------------------------------
[ 0, 99] |
[100, 199] | job x starts at 100, stops at 199, takes 100
[200, 299] |
```
after this patch, it's still the same:
```
batch | job
-----------------------------------------
[ 0, 99] |
[100, 199] | job y starts at 100, stops at 199, takes 100
[200, 299] |
```
--
@zsxwing thoughts on the above [1] and [2]? Thanks! :-)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/12797
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61774365
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/StreamTest.scala ---
@@ -280,19 +283,35 @@ trait StreamTest extends QueryTest with Timeouts {
try {
startedTest.foreach { action =>
action match {
- case StartStream =>
+ case StartStream(_trigger, _triggerClock) =>
verify(currentStream == null, "stream already running")
lastStream = currentStream
currentStream =
- sqlContext
- .streams
- .startQuery(
- StreamExecution.nextName,
- metadataRoot,
- stream,
- sink,
- outputMode = outputMode)
- .asInstanceOf[StreamExecution]
+ if (_trigger != null) {
--- End diff --
This is minor, but why the `_underscores` here? Its also a little confusing that we have these layers of default arguments. Should we get rid of the defaults in `startQuery` and only have them in this test code?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-216704764
LGTM pending tests.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61837747
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ProcessingTimeExecutorSuite.scala ---
@@ -21,19 +21,41 @@ import java.util.concurrent.{CountDownLatch, TimeUnit}
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.ProcessingTime
-import org.apache.spark.util.ManualClock
+import org.apache.spark.util.{Clock, ManualClock, SystemClock}
class ProcessingTimeExecutorSuite extends SparkFunSuite {
test("nextBatchTime") {
val processingTimeExecutor = ProcessingTimeExecutor(ProcessingTime(100))
+ assert(processingTimeExecutor.nextBatchTime(0) === 100)
assert(processingTimeExecutor.nextBatchTime(1) === 100)
assert(processingTimeExecutor.nextBatchTime(99) === 100)
- assert(processingTimeExecutor.nextBatchTime(100) === 100)
+ assert(processingTimeExecutor.nextBatchTime(100) === 200)
assert(processingTimeExecutor.nextBatchTime(101) === 200)
assert(processingTimeExecutor.nextBatchTime(150) === 200)
}
+ private def testNextBatchTimeAgainstClock(clock: Clock) {
+ val IntervalMS = 100
--- End diff --
Sure; let me fix this typo
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-215929409
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61663843
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/streaming/StreamSuite.scala ---
@@ -136,6 +136,22 @@ class StreamSuite extends StreamTest with SharedSQLContext {
testStream(ds)()
}
}
+
+ // This would fail for now -- error is "Timed out waiting for stream"
+ // Root cause is that data generated in batch 0 may not get processed in batch 1
+ // Let's enable this after SPARK-14942: Reduce delay between batch construction and execution
+ ignore("minimize delay between batch construction and execution") {
+ val inputData = MemoryStream[Int]
+ testStream(inputData.toDS())(
+ StartStream(ProcessingTime("10 seconds"), new ManualClock),
+ /* -- batch 0 ----------------------- */
+ AddData(inputData, 1),
+ AddData(inputData, 2),
+ AddData(inputData, 3),
+ AdvanceManualClock(10 * 1000), // 10 seconds
+ /* -- batch 1 ----------------------- */
+ CheckAnswer(1, 2, 3))
+ }
--- End diff --
The above test takes advantage of the new `StartStream` and `AdvanceManualClock` action.
It's testing against `SPARK-14942: Reduce delay between batch construction and execution` with a manually timed executor.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-216714401
**[Test build #57691 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/57691/consoleFull)** for PR 12797 at commit [`bc89962`](https://github.com/apache/spark/commit/bc899628fa67acfbe6bb3d8e2c0ba2aefc2422a6).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61976675
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala ---
@@ -65,8 +65,13 @@ case class ProcessingTimeExecutor(processingTime: ProcessingTime, clock: Clock =
s"${intervalMs} milliseconds, but spent ${realElapsedTimeMs} milliseconds")
}
- /** Return the next multiple of intervalMs */
+ /**
+ * Returns the start time in milliseconds for the next batch interval, given the current time.
+ * Note that a batch interval is inclusive with respect to its start time, and thus calling
+ * `nextBatchTime` with the result of a previous call should return the next interval. (i.e. given
+ * an interval of `100 ms`, `nextBatchTime(nextBatchTime(0)) = 200` rather than `0`).
--- End diff --
`nextBatchTime(0) = 100`, so `nextBatchTime(nextBatchTime(0)) = 200`?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-215925062
**[Test build #57391 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/57391/consoleFull)** for PR 12797 at commit [`9d80b15`](https://github.com/apache/spark/commit/9d80b15e33151f5f87207d72f89f016c18a21b01).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by marmbrus <gi...@git.apache.org>.
Github user marmbrus commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61773617
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala ---
@@ -65,8 +65,22 @@ case class ProcessingTimeExecutor(processingTime: ProcessingTime, clock: Clock =
s"${intervalMs} milliseconds, but spent ${realElapsedTimeMs} milliseconds")
}
- /** Return the next multiple of intervalMs */
+ /** Return the next multiple of intervalMs
+ *
+ * e.g. for intervalMs = 100
+ * nextBatchTime(0) = 100
+ * nextBatchTime(1) = 100
+ * ...
+ * nextBatchTime(99) = 100
+ * nextBatchTime(100) = 200
+ * nextBatchTime(101) = 200
+ * ...
+ * nextBatchTime(199) = 200
+ * nextBatchTime(200) = 300
+ *
+ * Note, this way, we'll get nextBatchTime(nextBatchTime(0)) = 200, rather than = 0
--- End diff --
This comment took me a while to understand, what do you think about this?
```
/**
* Returns the start time in milliseconds for the next batch interval, given the current time.
* Note that a batch interval is inclusive with respect to its start time, and thus calling
* `nextBatchTime` with the result of a previous call should return the next interval. (i.e. given
* an interval of `100 ms`, `nextBatchTime(nextBatchTime(0)) = 200` rather than `0`).
*/
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-216714507
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61977204
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala ---
@@ -65,8 +65,13 @@ case class ProcessingTimeExecutor(processingTime: ProcessingTime, clock: Clock =
s"${intervalMs} milliseconds, but spent ${realElapsedTimeMs} milliseconds")
}
- /** Return the next multiple of intervalMs */
+ /**
+ * Returns the start time in milliseconds for the next batch interval, given the current time.
+ * Note that a batch interval is inclusive with respect to its start time, and thus calling
+ * `nextBatchTime` with the result of a previous call should return the next interval. (i.e. given
+ * an interval of `100 ms`, `nextBatchTime(nextBatchTime(0)) = 200` rather than `0`).
--- End diff --
Thank you always for the prompt review!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-215929256
**[Test build #57391 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/57391/consoleFull)** for PR 12797 at commit [`9d80b15`](https://github.com/apache/spark/commit/9d80b15e33151f5f87207d72f89f016c18a21b01).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-215929410
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/57391/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-216701369
**[Test build #57691 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/57691/consoleFull)** for PR 12797 at commit [`bc89962`](https://github.com/apache/spark/commit/bc899628fa67acfbe6bb3d8e2c0ba2aefc2422a6).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-215928373
@marmbrus @tdas @zsxwing would you mind taking a look? Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61663787
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/execution/streaming/ProcessingTimeExecutorSuite.scala ---
@@ -21,19 +21,41 @@ import java.util.concurrent.{CountDownLatch, TimeUnit}
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.ProcessingTime
-import org.apache.spark.util.ManualClock
+import org.apache.spark.util.{Clock, ManualClock, SystemClock}
class ProcessingTimeExecutorSuite extends SparkFunSuite {
test("nextBatchTime") {
val processingTimeExecutor = ProcessingTimeExecutor(ProcessingTime(100))
+ assert(processingTimeExecutor.nextBatchTime(0) === 100)
assert(processingTimeExecutor.nextBatchTime(1) === 100)
assert(processingTimeExecutor.nextBatchTime(99) === 100)
- assert(processingTimeExecutor.nextBatchTime(100) === 100)
+ assert(processingTimeExecutor.nextBatchTime(100) === 200)
assert(processingTimeExecutor.nextBatchTime(101) === 200)
assert(processingTimeExecutor.nextBatchTime(150) === 200)
}
+ private def testNextBatchTimeAgainstClock(clock: Clock) {
+ val IntervalMS = 100
+ val processingTimeExecutor = ProcessingTimeExecutor(ProcessingTime(IntervalMS), clock)
+
+ val ITERATION = 10
+ var nextBatchTime: Long = 0
+ for (it <- 1 to ITERATION)
+ nextBatchTime = processingTimeExecutor.nextBatchTime(nextBatchTime)
+
+ // nextBatchTime should be 1000
+ assert(nextBatchTime === IntervalMS * ITERATION)
+ }
+
+ test("nextBatchTime against SystemClock") {
+ testNextBatchTimeAgainstClock(new SystemClock)
+ }
+
+ test("nextBatchTime against ManualClock") {
--- End diff --
Please note the `ProcessingTimeExecutor` issue would fail this test without this patch, but would pass with this patch.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61775300
--- Diff: sql/core/src/test/scala/org/apache/spark/sql/StreamTest.scala ---
@@ -142,7 +142,10 @@ trait StreamTest extends QueryTest with Timeouts {
case object StopStream extends StreamAction with StreamMustBeRunning
/** Starts the stream, resuming if data has already been processed. It must not be running. */
- case object StartStream extends StreamAction
+ case class StartStream(trigger: Trigger = null, triggerClock: Clock = null) extends StreamAction
--- End diff --
You can use the same default values of `StreamExecution`. Then you don't need to handle the `null` case.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61977039
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala ---
@@ -65,8 +65,13 @@ case class ProcessingTimeExecutor(processingTime: ProcessingTime, clock: Clock =
s"${intervalMs} milliseconds, but spent ${realElapsedTimeMs} milliseconds")
}
- /** Return the next multiple of intervalMs */
+ /**
+ * Returns the start time in milliseconds for the next batch interval, given the current time.
+ * Note that a batch interval is inclusive with respect to its start time, and thus calling
+ * `nextBatchTime` with the result of a previous call should return the next interval. (i.e. given
+ * an interval of `100 ms`, `nextBatchTime(nextBatchTime(0)) = 200` rather than `0`).
--- End diff --
Oh, right. Sorry for the mistake.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023][SQL][Streaming] Add...
Posted by zsxwing <gi...@git.apache.org>.
Github user zsxwing commented on a diff in the pull request:
https://github.com/apache/spark/pull/12797#discussion_r61976436
--- Diff: sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/TriggerExecutor.scala ---
@@ -65,8 +65,13 @@ case class ProcessingTimeExecutor(processingTime: ProcessingTime, clock: Clock =
s"${intervalMs} milliseconds, but spent ${realElapsedTimeMs} milliseconds")
}
- /** Return the next multiple of intervalMs */
+ /**
+ * Returns the start time in milliseconds for the next batch interval, given the current time.
+ * Note that a batch interval is inclusive with respect to its start time, and thus calling
+ * `nextBatchTime` with the result of a previous call should return the next interval. (i.e. given
+ * an interval of `100 ms`, `nextBatchTime(nextBatchTime(0)) = 200` rather than `0`).
--- End diff --
nit: `nextBatchTime(nextBatchTime(0)) = 200` -> `nextBatchTime(nextBatchTime(100)) = 200`
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-15022][SPARK-15023] Add support for tes...
Posted by lw-lin <gi...@git.apache.org>.
Github user lw-lin commented on the pull request:
https://github.com/apache/spark/pull/12797#issuecomment-215924685
still editing, will explain why ProcessingTimeExecutor, where for a batch it should run batchRunner only once but might run multiple times under certain conditions;
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org