[SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doesn't work all the time #5871

wesleymiao · 2015-05-03T12:21:20Z

https://issues.apache.org/jira/browse/SPARK-7326

The problem most likely resides in DStream.slice() implementation, as shown below.

def slice(fromTime: Time, toTime: Time): Seq[RDD[T]] = {
if (!isInitialized) {
throw new SparkException(this + " has not been initialized")
}
if (!(fromTime - zeroTime).isMultipleOf(slideDuration)) {
logWarning("fromTime (" + fromTime + ") is not a multiple of slideDuration ("
+ slideDuration + ")")
}
if (!(toTime - zeroTime).isMultipleOf(slideDuration)) {
logWarning("toTime (" + fromTime + ") is not a multiple of slideDuration ("
+ slideDuration + ")")
}
val alignedToTime = toTime.floor(slideDuration, zeroTime)
val alignedFromTime = fromTime.floor(slideDuration, zeroTime)

logInfo("Slicing from " + fromTime + " to " + toTime +
  " (aligned to " + alignedFromTime + " and " + alignedToTime + ")")

alignedFromTime.to(alignedToTime, slideDuration).flatMap(time => {
  if (time >= zeroTime) getOrCompute(time) else None
})

}

Here after performing floor() on both fromTime and toTime, the result (alignedFromTime - zeroTime) and (alignedToTime - zeroTime) may no longer be multiple of the slidingDuration, thus making isTimeValid() check failed for all the remaining computation.

The fix is to add a new floor() function in Time.scala to respect the zeroTime while performing the floor :

def floor(that: Duration, zeroTime: Time): Time = {
val t = that.milliseconds
new Time(((this.millis - zeroTime.milliseconds) / t) * t + zeroTime.milliseconds)
}

And then change the DStream.slice to call this new floor function by passing in its zeroTime.

val alignedToTime = toTime.floor(slideDuration, zeroTime)
val alignedFromTime = fromTime.floor(slideDuration, zeroTime)

This way the alignedToTime and alignedFromTime are really aligned in respect to zeroTime whose value is not really a 0.

…sn't work all the time The problem most likely resides in DStream.slice() implementation, as shown below. def slice(fromTime: Time, toTime: Time): Seq[RDD[T]] = { if (!isInitialized) { throw new SparkException(this + " has not been initialized") } if (!(fromTime - zeroTime).isMultipleOf(slideDuration)) { logWarning("fromTime (" + fromTime + ") is not a multiple of slideDuration (" + slideDuration + ")") } if (!(toTime - zeroTime).isMultipleOf(slideDuration)) { logWarning("toTime (" + fromTime + ") is not a multiple of slideDuration (" + slideDuration + ")") } val alignedToTime = toTime.floor(slideDuration, zeroTime) val alignedFromTime = fromTime.floor(slideDuration, zeroTime) logInfo("Slicing from " + fromTime + " to " + toTime + " (aligned to " + alignedFromTime + " and " + alignedToTime + ")") alignedFromTime.to(alignedToTime, slideDuration).flatMap(time => { if (time >= zeroTime) getOrCompute(time) else None }) } Here after performing floor() on both fromTime and toTime, the result (alignedFromTime - zeroTime) and (alignedToTime - zeroTime) may no longer be multiple of the slidingDuration, thus making isTimeValid() check failed for all the remaining computation. The fix is to add a new floor() function in Time.scala to respect the zeroTime while performing the floor : def floor(that: Duration, zeroTime: Time): Time = { val t = that.milliseconds new Time(((this.millis - zeroTime.milliseconds) / t) * t + zeroTime.milliseconds) } And then change the DStream.slice to call this new floor function by passing in its zeroTime. val alignedToTime = toTime.floor(slideDuration, zeroTime) val alignedFromTime = fromTime.floor(slideDuration, zeroTime) This way the alignedToTime and alignedFromTime are *really* aligned in respect to zeroTime whose value is not really a 0.

AmplabJenkins · 2015-05-03T12:22:10Z

Can one of the admins verify this patch?

srowen · 2015-05-03T12:47:32Z

The intent of the code does seem to be that the difference from zeroTime is what is supposed to be a multiple of slideDuration. From my understanding I agree with your change. Let's see what the tests say.

srowen · 2015-05-03T12:47:36Z

ok to test

AmplabJenkins · 2015-05-03T12:52:09Z

Merged build triggered.

AmplabJenkins · 2015-05-03T12:52:19Z

Merged build started.

SparkQA · 2015-05-03T12:53:52Z

Test build #31691 has started for PR 5871 at commit 2611745.

SparkQA · 2015-05-03T14:36:50Z

Test build #31691 has finished for PR 5871 at commit 2611745.

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

viirya · 2015-05-03T14:36:52Z

You will not directly call slice(). When slice() is called internally, e.g. in WindowedDStream.compute(), the Interval given as parameter to slice is valid. So if it is not trying to directly call slice, calling floor() to align fromTime and toTime is unnecessary.

As you pointed, the current floor is buggy. That is why your example runs incorrectly. In fact, just commenting out two floor callings makes your example passed.

Is it better to only perform alignment when the isMultipleOf checking is failed?

AmplabJenkins · 2015-05-03T14:36:56Z

Merged build finished. Test PASSed.

AmplabJenkins · 2015-05-03T14:36:56Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/31691/
Test PASSed.

…sn't work all the time Improve it a bit to only do floor when it is not aligned relative to its zeroTime

AmplabJenkins · 2015-05-04T00:37:10Z

Merged build triggered.

AmplabJenkins · 2015-05-04T00:37:17Z

Merged build started.

SparkQA · 2015-05-04T00:38:53Z

Test build #31715 has started for PR 5871 at commit 6ade399.

wesleymiao · 2015-05-04T00:39:03Z

@viirya , you are right. I've changed a little bit to only perform floor when the fromTime and toTime are not aligned relative to zeroTime.

Also fixed a typo in the log message.

SparkQA · 2015-05-04T02:21:16Z

Test build #31715 has finished for PR 5871 at commit 6ade399.

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

AmplabJenkins · 2015-05-04T02:21:21Z

Merged build finished. Test PASSed.

AmplabJenkins · 2015-05-04T02:21:21Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/31715/
Test PASSed.

srowen · 2015-05-04T05:38:42Z

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

-      logWarning("fromTime (" + fromTime + ") is not a multiple of slideDuration ("
-        + slideDuration + ")")
+
+    val alignedToTime = (toTime - zeroTime).isMultipleOf(slideDuration) match {


I think match syntax for booleans is needlessly complicated. You can just write a more straightforward if.

changed to if-else.

…sn't work all the time In response to comment, change match-case to if-else for boolean to be consistent.

AmplabJenkins · 2015-05-04T06:12:09Z

Merged build triggered.

AmplabJenkins · 2015-05-04T06:12:19Z

Merged build started.

SparkQA · 2015-05-04T06:13:50Z

Test build #31730 has started for PR 5871 at commit 48b4dc0.

SparkQA · 2015-05-04T07:55:53Z

Test build #31730 has finished for PR 5871 at commit 48b4dc0.

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

AmplabJenkins · 2015-05-04T07:55:57Z

Merged build finished. Test PASSed.

AmplabJenkins · 2015-05-04T07:55:58Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/31730/
Test PASSed.

zsxwing · 2015-05-05T20:44:21Z

streaming/src/main/scala/org/apache/spark/streaming/Time.scala

@@ -63,6 +63,11 @@ case class Time(private val millis: Long) {
    new Time((this.millis / t) * t)
  }

+  def floor(that: Duration, zeroTime: Time): Time = {


Could you also add some tests for this new method?

Added a few tests.

Should this be made public? If so, would be good to add some docs.

srowen · 2015-05-08T08:34:51Z

A little test would be great, yeah. Would still love a quick look from @tdas to make sure we're not missing something.

…en't work all the time Add some tests to cover the new function Time.floor(duration, zeroTime)

AmplabJenkins · 2015-05-08T13:27:11Z

Merged build triggered.

AmplabJenkins · 2015-05-08T13:27:18Z

Merged build started.

SparkQA · 2015-05-08T13:28:59Z

Test build #32230 has started for PR 5871 at commit 82a4d8c.

AmplabJenkins · 2015-05-08T14:36:08Z

Merged build finished. Test FAILed.

AmplabJenkins · 2015-05-08T14:36:08Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/32230/
Test FAILed.

shaneknapp · 2015-05-08T14:36:36Z

i'll retrigger this job once i restart jenkins

shaneknapp · 2015-05-08T14:41:10Z

jenkins, test this please

AmplabJenkins · 2015-05-08T14:42:36Z

Merged build triggered.

AmplabJenkins · 2015-05-08T14:42:41Z

Merged build started.

SparkQA · 2015-05-08T14:44:34Z

Test build #32237 has started for PR 5871 at commit 82a4d8c.

SparkQA · 2015-05-08T16:29:50Z

Test build #32237 has finished for PR 5871 at commit 82a4d8c.

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

AmplabJenkins · 2015-05-08T16:29:55Z

Merged build finished. Test PASSed.

AmplabJenkins · 2015-05-08T16:29:56Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/32237/
Test PASSed.

wesleymiao · 2015-05-08T23:22:13Z

Hi, @tdas , will you be able to take a look at this PR? My team in Autodesk is trying to adopt Spark Streaming to do some real-time analytics for our services' API logs to serve various purposes, among which is to monitor our services' health in real-time based on API log statistics. Then we run into this situation where we need to perform reduceByKeyAndWindow on an already windowed DStream.

What I'd like to achieve is to do multiple-level reduce on the source of API logging stream. For example - let's say the source logging stream is at interval 1 second. The first level aggregation would be every 1 minute, which is the 60 intervals of the source stream. The second level would be every 1 hour. The third level would be 1 day. And we can do more levels if we want.

What I hope is that at each level we'll do reduceByKeyAndWindow so that its aggregation can be done over its immediate previous level, instead of always aggregating over the source stream. Level 3's reduceByKey will be based on level 2's result, level 2 is based one level 1 and level 1 is based on the source stream.

I would think this approach will be more efficient than always reducing over the source stream, particularly for the higher level (like daily and weekly) aggregation.

srowen · 2015-05-11T11:19:39Z

I'm proceeding with this since everywhere in the code it appears that the intent is for RDDs to start an end on a multiple of the slide duration after zero time. Therefore it seems like the intent must have been to take into account zeroTime.

…sn't work all the time tdas https://issues.apache.org/jira/browse/SPARK-7326 The problem most likely resides in DStream.slice() implementation, as shown below. def slice(fromTime: Time, toTime: Time): Seq[RDD[T]] = { if (!isInitialized) { throw new SparkException(this + " has not been initialized") } if (!(fromTime - zeroTime).isMultipleOf(slideDuration)) { logWarning("fromTime (" + fromTime + ") is not a multiple of slideDuration (" + slideDuration + ")") } if (!(toTime - zeroTime).isMultipleOf(slideDuration)) { logWarning("toTime (" + fromTime + ") is not a multiple of slideDuration (" + slideDuration + ")") } val alignedToTime = toTime.floor(slideDuration, zeroTime) val alignedFromTime = fromTime.floor(slideDuration, zeroTime) logInfo("Slicing from " + fromTime + " to " + toTime + " (aligned to " + alignedFromTime + " and " + alignedToTime + ")") alignedFromTime.to(alignedToTime, slideDuration).flatMap(time => { if (time >= zeroTime) getOrCompute(time) else None }) } Here after performing floor() on both fromTime and toTime, the result (alignedFromTime - zeroTime) and (alignedToTime - zeroTime) may no longer be multiple of the slidingDuration, thus making isTimeValid() check failed for all the remaining computation. The fix is to add a new floor() function in Time.scala to respect the zeroTime while performing the floor : def floor(that: Duration, zeroTime: Time): Time = { val t = that.milliseconds new Time(((this.millis - zeroTime.milliseconds) / t) * t + zeroTime.milliseconds) } And then change the DStream.slice to call this new floor function by passing in its zeroTime. val alignedToTime = toTime.floor(slideDuration, zeroTime) val alignedFromTime = fromTime.floor(slideDuration, zeroTime) This way the alignedToTime and alignedFromTime are *really* aligned in respect to zeroTime whose value is not really a 0. Author: Wesley Miao <wesley.miao@gmail.com> Author: Wesley <wesley.miao@autodesk.com> Closes #5871 from wesleymiao/spark-7326 and squashes the following commits: 82a4d8c [Wesley Miao] [SPARK-7326] [STREAMING] Performing window() on a WindowedDStream dosen't work all the time 48b4dc0 [Wesley] [SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doesn't work all the time 6ade399 [Wesley] [SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doesn't work all the time 2611745 [Wesley Miao] [SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doesn't work all the time (cherry picked from commit d70a076) Signed-off-by: Sean Owen <sowen@cloudera.com>

tdas · 2015-05-11T18:48:21Z

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

+      fromTime
+    } else {
+      logWarning("fromTime (" + fromTime + ") is not a multiple of slideDuration ("
+      + slideDuration + ")")


Incorrect indentation.

tdas · 2015-05-11T18:50:01Z

Aah, this has already been merged before I got to it. That's okay, barring a few nits, it was good. Thanks @wesleymiao @srowen

…sn't work all the time tdas https://issues.apache.org/jira/browse/SPARK-7326 The problem most likely resides in DStream.slice() implementation, as shown below. def slice(fromTime: Time, toTime: Time): Seq[RDD[T]] = { if (!isInitialized) { throw new SparkException(this + " has not been initialized") } if (!(fromTime - zeroTime).isMultipleOf(slideDuration)) { logWarning("fromTime (" + fromTime + ") is not a multiple of slideDuration (" + slideDuration + ")") } if (!(toTime - zeroTime).isMultipleOf(slideDuration)) { logWarning("toTime (" + fromTime + ") is not a multiple of slideDuration (" + slideDuration + ")") } val alignedToTime = toTime.floor(slideDuration, zeroTime) val alignedFromTime = fromTime.floor(slideDuration, zeroTime) logInfo("Slicing from " + fromTime + " to " + toTime + " (aligned to " + alignedFromTime + " and " + alignedToTime + ")") alignedFromTime.to(alignedToTime, slideDuration).flatMap(time => { if (time >= zeroTime) getOrCompute(time) else None }) } Here after performing floor() on both fromTime and toTime, the result (alignedFromTime - zeroTime) and (alignedToTime - zeroTime) may no longer be multiple of the slidingDuration, thus making isTimeValid() check failed for all the remaining computation. The fix is to add a new floor() function in Time.scala to respect the zeroTime while performing the floor : def floor(that: Duration, zeroTime: Time): Time = { val t = that.milliseconds new Time(((this.millis - zeroTime.milliseconds) / t) * t + zeroTime.milliseconds) } And then change the DStream.slice to call this new floor function by passing in its zeroTime. val alignedToTime = toTime.floor(slideDuration, zeroTime) val alignedFromTime = fromTime.floor(slideDuration, zeroTime) This way the alignedToTime and alignedFromTime are *really* aligned in respect to zeroTime whose value is not really a 0. Author: Wesley Miao <wesley.miao@gmail.com> Author: Wesley <wesley.miao@autodesk.com> Closes apache#5871 from wesleymiao/spark-7326 and squashes the following commits: 82a4d8c [Wesley Miao] [SPARK-7326] [STREAMING] Performing window() on a WindowedDStream dosen't work all the time 48b4dc0 [Wesley] [SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doesn't work all the time 6ade399 [Wesley] [SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doesn't work all the time 2611745 [Wesley Miao] [SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doesn't work all the time

[SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doe…

6ade399

…sn't work all the time Improve it a bit to only do floor when it is not aligned relative to its zeroTime

srowen reviewed May 4, 2015
View reviewed changes

[SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doe…

48b4dc0

…sn't work all the time In response to comment, change match-case to if-else for boolean to be consistent.

zsxwing reviewed May 5, 2015
View reviewed changes

[SPARK-7326] [STREAMING] Performing window() on a WindowedDStream dos…

82a4d8c

…en't work all the time Add some tests to cover the new function Time.floor(duration, zeroTime)

asfgit closed this in d70a076 May 11, 2015

tdas reviewed May 11, 2015
View reviewed changes

[SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doesn't work all the time #5871

[SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doesn't work all the time #5871

Conversation

wesleymiao commented May 3, 2015

AmplabJenkins commented May 3, 2015

srowen commented May 3, 2015

srowen commented May 3, 2015

AmplabJenkins commented May 3, 2015

AmplabJenkins commented May 3, 2015

SparkQA commented May 3, 2015

SparkQA commented May 3, 2015

viirya commented May 3, 2015

AmplabJenkins commented May 3, 2015

AmplabJenkins commented May 3, 2015

AmplabJenkins commented May 4, 2015

AmplabJenkins commented May 4, 2015

SparkQA commented May 4, 2015

wesleymiao commented May 4, 2015

SparkQA commented May 4, 2015

AmplabJenkins commented May 4, 2015

AmplabJenkins commented May 4, 2015

srowen May 4, 2015

Choose a reason for hiding this comment

wesleymiao May 4, 2015

Choose a reason for hiding this comment

AmplabJenkins commented May 4, 2015

AmplabJenkins commented May 4, 2015

SparkQA commented May 4, 2015

SparkQA commented May 4, 2015

AmplabJenkins commented May 4, 2015

AmplabJenkins commented May 4, 2015

zsxwing May 5, 2015

Choose a reason for hiding this comment

wesleymiao May 8, 2015

Choose a reason for hiding this comment

tdas May 11, 2015

Choose a reason for hiding this comment

srowen commented May 8, 2015

AmplabJenkins commented May 8, 2015

AmplabJenkins commented May 8, 2015

SparkQA commented May 8, 2015

AmplabJenkins commented May 8, 2015

AmplabJenkins commented May 8, 2015

shaneknapp commented May 8, 2015

shaneknapp commented May 8, 2015

AmplabJenkins commented May 8, 2015

AmplabJenkins commented May 8, 2015

SparkQA commented May 8, 2015

SparkQA commented May 8, 2015

AmplabJenkins commented May 8, 2015

AmplabJenkins commented May 8, 2015

wesleymiao commented May 8, 2015

srowen commented May 11, 2015

tdas May 11, 2015

Choose a reason for hiding this comment

tdas commented May 11, 2015