{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/15702","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/15702/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/15702/comments","events_url":"https://api.github.com/repos/apache/spark/issues/15702/events","html_url":"https://github.com/apache/spark/pull/15702","id":186414721,"number":15702,"title":"[SPARK-18124] Observed delay based Event Time Watermarks","user":{"login":"marmbrus","id":527,"avatar_url":"https://avatars.githubusercontent.com/u/527?v=3","gravatar_id":"","url":"https://api.github.com/users/marmbrus","html_url":"https://github.com/marmbrus","followers_url":"https://api.github.com/users/marmbrus/followers","following_url":"https://api.github.com/users/marmbrus/following{/other_user}","gists_url":"https://api.github.com/users/marmbrus/gists{/gist_id}","starred_url":"https://api.github.com/users/marmbrus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/marmbrus/subscriptions","organizations_url":"https://api.github.com/users/marmbrus/orgs","repos_url":"https://api.github.com/users/marmbrus/repos","events_url":"https://api.github.com/users/marmbrus/events{/privacy}","received_events_url":"https://api.github.com/users/marmbrus/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":6,"created_at":"2016-10-31T22:22:07Z","updated_at":"2016-11-01T19:09:34Z","closed_at":null,"pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/15702","html_url":"https://github.com/apache/spark/pull/15702","diff_url":"https://github.com/apache/spark/pull/15702.diff","patch_url":"https://github.com/apache/spark/pull/15702.patch"},"body":"This PR adds a new method `withWatermark` to the `Dataset` API, which can be used specify an _event time watermark_.  An event time watermark allows the streaming engine to reason about the point in time after which we no longer expect to see late data.  This PR also has augmented `StreamExecution` to use this watermark for several purposes:\r\n  - To know when a given time window aggregation is finalized and thus results can be emitted when using output modes that do not allow updates (e.g. `Append` mode).\r\n  - To minimize the amount of state that we need to keep for on-going aggregations, by evicting state for groups that are no longer expected to change.  Although, we do still maintain all state if the query requires (i.e. if the event time is not present in the `groupBy` or when running in `Complete` mode).\r\n\r\nAn example that emits windowed counts of records, waiting up to 5 minutes for late data to arrive.\r\n```scala\r\ndf.withWatermark(\"eventTime\", \"5 minutes\")\r\n  .groupBy(window($\"eventTime\", \"1 minute\") as 'window)\r\n  .count()\r\n  .writeStream\r\n  .format(\"console\")\r\n  .mode(\"append\") // In append mode, we only output finalized aggregations.\r\n  .start()\r\n```\r\n\r\n### Calculating the watermark.\r\nThe current event time is computed by looking at the `MAX(eventTime)` seen this epoch across all of the partitions in the query minus some user defined _delayThreshold_.  An additional constraint is that the watermark must increase monotonically.\r\n\r\nNote that since we must coordinate this value across partitions occasionally, the actual watermark used is only guaranteed to be at least `delay` behind the actual event time.  In some cases we may still process records that arrive more than delay late.\r\n\r\nThis mechanism was chosen for the initial implementation over processing time for two reasons:   \r\n  - it is robust to downtime that could affect processing delay\r\n  - it does not require syncing of time or timezones between the producer and the processing engine.\r\n\r\n### Other notable implementation details\r\n - A new trigger metric `eventTimeWatermark` outputs the current value of the watermark.\r\n - We mark the event time column in the `Attribute` metadata using the key `spark.watermarkDelay`.  This allows downstream operations to know which column holds the event time.  Operations like `window` propagate this metadata.\r\n - `explain()` marks the watermark with a suffix of `-T${delayMs}` to ease debugging of how this information is propagated.\r\n - Currently, we don't filter out late records, but instead rely on the state store to avoid emitting records that are both added and filtered in the same epoch.\r\n\r\n### Remaining in this PR\r\n - [ ] The test for recovery is currently failing as we don't record the watermark used in the offset log.  We will need to do so to ensure determinism, but this is deferred until #15626 is merged.\r\n\r\n### Other follow-ups\r\nThere are some natural additional features that we should consider for future work:\r\n - Ability to write records that arrive too late to some external store in case any out-of-band remediation is required.\r\n - `Update` mode so you can get partial results before a group is evicted.\r\n - Other mechanisms for calculating the watermark.  In particular a watermark based on quantiles would be more robust to outliers."},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/257663607","html_url":"https://github.com/apache/spark/pull/15702#issuecomment-257663607","issue_url":"https://api.github.com/repos/apache/spark/issues/15702","id":257663607,"user":{"login":"marmbrus","id":527,"avatar_url":"https://avatars.githubusercontent.com/u/527?v=3","gravatar_id":"","url":"https://api.github.com/users/marmbrus","html_url":"https://github.com/marmbrus","followers_url":"https://api.github.com/users/marmbrus/followers","following_url":"https://api.github.com/users/marmbrus/following{/other_user}","gists_url":"https://api.github.com/users/marmbrus/gists{/gist_id}","starred_url":"https://api.github.com/users/marmbrus/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/marmbrus/subscriptions","organizations_url":"https://api.github.com/users/marmbrus/orgs","repos_url":"https://api.github.com/users/marmbrus/repos","events_url":"https://api.github.com/users/marmbrus/events{/privacy}","received_events_url":"https://api.github.com/users/marmbrus/received_events","type":"User","site_admin":false},"created_at":"2016-11-01T19:09:34Z","updated_at":"2016-11-01T19:09:34Z","body":"Not a dumb question!  You can certainly use processing time if those are the semantics you require.  I do think there is a little bit of work we need to do to ensure determinism for these functions.  Specifically, `now()` in SQL is supposed to return the moment query execution began.  For streaming we'll need to record this and make sure that we give the same time even in the case of failures (based on the batch id with the current execution model).\r\n\r\nGood point on the documentation.  The thing you are missing is that it must be used in a window function, which does require `TimestampType`.  I can see how to make this more clear."}}
{"actor":{"display_login":"marmbrus"}}