{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/5871","labels_url":"https://api.github.com/repos/apache/spark/issues/5871/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/5871/comments","events_url":"https://api.github.com/repos/apache/spark/issues/5871/events","html_url":"https://github.com/apache/spark/pull/5871","id":72798023,"number":5871,"title":"[SPARK-7326] [STREAMING] Performing window() on a WindowedDStream doesn't work all the time","user":{"login":"wesleymiao","id":12214644,"avatar_url":"https://avatars.githubusercontent.com/u/12214644?v=3","gravatar_id":"","url":"https://api.github.com/users/wesleymiao","html_url":"https://github.com/wesleymiao","followers_url":"https://api.github.com/users/wesleymiao/followers","following_url":"https://api.github.com/users/wesleymiao/following{/other_user}","gists_url":"https://api.github.com/users/wesleymiao/gists{/gist_id}","starred_url":"https://api.github.com/users/wesleymiao/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/wesleymiao/subscriptions","organizations_url":"https://api.github.com/users/wesleymiao/orgs","repos_url":"https://api.github.com/users/wesleymiao/repos","events_url":"https://api.github.com/users/wesleymiao/events{/privacy}","received_events_url":"https://api.github.com/users/wesleymiao/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":26,"created_at":"2015-05-03T12:21:20Z","updated_at":"2015-05-08T13:27:18Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/5871","html_url":"https://github.com/apache/spark/pull/5871","diff_url":"https://github.com/apache/spark/pull/5871.diff","patch_url":"https://github.com/apache/spark/pull/5871.patch"},"body":"@tdas \r\n\r\nThe problem most likely resides in DStream.slice() implementation, as shown below.\r\n\r\n  def slice(fromTime: Time, toTime: Time): Seq[RDD[T]] = {\r\n    if (!isInitialized) {\r\n      throw new SparkException(this + \" has not been initialized\")\r\n    }\r\n    if (!(fromTime - zeroTime).isMultipleOf(slideDuration)) {\r\n      logWarning(\"fromTime (\" + fromTime + \") is not a multiple of slideDuration (\"\r\n        + slideDuration + \")\")\r\n    }\r\n    if (!(toTime - zeroTime).isMultipleOf(slideDuration)) {\r\n      logWarning(\"toTime (\" + fromTime + \") is not a multiple of slideDuration (\"\r\n        + slideDuration + \")\")\r\n    }\r\n    val alignedToTime = toTime.floor(slideDuration, zeroTime)\r\n    val alignedFromTime = fromTime.floor(slideDuration, zeroTime)\r\n\r\n    logInfo(\"Slicing from \" + fromTime + \" to \" + toTime +\r\n      \" (aligned to \" + alignedFromTime + \" and \" + alignedToTime + \")\")\r\n\r\n    alignedFromTime.to(alignedToTime, slideDuration).flatMap(time => {\r\n      if (time >= zeroTime) getOrCompute(time) else None\r\n    })\r\n  }\r\n\r\nHere 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.\r\n\r\nThe fix is to add a new floor() function in Time.scala to respect the zeroTime while performing the floor :\r\n\r\n  def floor(that: Duration, zeroTime: Time): Time = {\r\n    val t = that.milliseconds\r\n    new Time(((this.millis - zeroTime.milliseconds) / t) * t + zeroTime.milliseconds)\r\n  }\r\n\r\nAnd then change the DStream.slice to call this new floor function by passing in its zeroTime.\r\n\r\n    val alignedToTime = toTime.floor(slideDuration, zeroTime)\r\n    val alignedFromTime = fromTime.floor(slideDuration, zeroTime)\r\n\r\nThis way the alignedToTime and alignedFromTime are *really* aligned in respect to zeroTime whose value is not really a 0."},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/100233275","html_url":"https://github.com/apache/spark/pull/5871#issuecomment-100233275","issue_url":"https://api.github.com/repos/apache/spark/issues/5871","id":100233275,"user":{"login":"AmplabJenkins","id":4031291,"avatar_url":"https://avatars.githubusercontent.com/u/4031291?v=3","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2015-05-08T13:27:18Z","updated_at":"2015-05-08T13:27:18Z","body":"Merged build started."}}