{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/23576","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/23576/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/23576/comments","events_url":"https://api.github.com/repos/apache/spark/issues/23576/events","html_url":"https://github.com/apache/spark/pull/23576","id":400468557,"node_id":"MDExOlB1bGxSZXF1ZXN0MjQ1NjQ5NzE3","number":23576,"title":"[SPARK-26655] [SS] Support multiple aggregates in append mode","user":{"login":"arunmahadevan","id":6792890,"node_id":"MDQ6VXNlcjY3OTI4OTA=","avatar_url":"https://avatars0.githubusercontent.com/u/6792890?v=4","gravatar_id":"","url":"https://api.github.com/users/arunmahadevan","html_url":"https://github.com/arunmahadevan","followers_url":"https://api.github.com/users/arunmahadevan/followers","following_url":"https://api.github.com/users/arunmahadevan/following{/other_user}","gists_url":"https://api.github.com/users/arunmahadevan/gists{/gist_id}","starred_url":"https://api.github.com/users/arunmahadevan/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/arunmahadevan/subscriptions","organizations_url":"https://api.github.com/users/arunmahadevan/orgs","repos_url":"https://api.github.com/users/arunmahadevan/repos","events_url":"https://api.github.com/users/arunmahadevan/events{/privacy}","received_events_url":"https://api.github.com/users/arunmahadevan/received_events","type":"User","site_admin":false},"labels":[{"id":1406587328,"node_id":"MDU6TGFiZWwxNDA2NTg3MzI4","url":"https://api.github.com/repos/apache/spark/labels/STRUCTURED%20STREAMING","name":"STRUCTURED STREAMING","color":"ededed","default":false}],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":29,"created_at":"2019-01-17T21:22:08Z","updated_at":"2019-08-27T09:19:50Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/23576","html_url":"https://github.com/apache/spark/pull/23576","diff_url":"https://github.com/apache/spark/pull/23576.diff","patch_url":"https://github.com/apache/spark/pull/23576.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis patch proposes to add support for multiple aggregates in append mode. In\r\nappend mode, the aggregates are emitted only after the watermark passes\r\nthe threshold (e.g. the window boundary) and the emitted value is not\r\naffected by further late data. This allows to chain multiple aggregates\r\nin 'Append' output mode without worrying about retractions etc.\r\n\r\nHowever the current event time watermarks in structured streaming are\r\ntracked at a global level and this does not work when aggregates are\r\nchained. The downstream watermarks usually lags the ones before and the\r\nglobal (min or max) watermarks will not let the stages make progress\r\nindependently.\r\n\r\nThe patch tracks the watermarks at each (stateful)\r\noperator so that the aggregate outputs are generated when the watermark\r\npasses the thresholds at the corresponding stateful operator. The values\r\nare also saved into the commit/offset logs (similar to global watermark)\r\n\r\nEach aggregate should have a corresponding watermark defined while\r\ncreating the query (E.g. via withWatermark) and this is used to\r\ntrack the progress of event time corresponding to the stateful operator.\r\n\r\n\r\n## How was this patch tested?\r\n\r\nNew and existing unit tests\r\n\r\nPlease review http://spark.apache.org/contributing.html before opening a pull request.\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/525217328","html_url":"https://github.com/apache/spark/pull/23576#issuecomment-525217328","issue_url":"https://api.github.com/repos/apache/spark/issues/23576","id":525217328,"node_id":"MDEyOklzc3VlQ29tbWVudDUyNTIxNzMyOA==","user":{"login":"HeartSaVioR","id":1317309,"node_id":"MDQ6VXNlcjEzMTczMDk=","avatar_url":"https://avatars2.githubusercontent.com/u/1317309?v=4","gravatar_id":"","url":"https://api.github.com/users/HeartSaVioR","html_url":"https://github.com/HeartSaVioR","followers_url":"https://api.github.com/users/HeartSaVioR/followers","following_url":"https://api.github.com/users/HeartSaVioR/following{/other_user}","gists_url":"https://api.github.com/users/HeartSaVioR/gists{/gist_id}","starred_url":"https://api.github.com/users/HeartSaVioR/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/HeartSaVioR/subscriptions","organizations_url":"https://api.github.com/users/HeartSaVioR/orgs","repos_url":"https://api.github.com/users/HeartSaVioR/repos","events_url":"https://api.github.com/users/HeartSaVioR/events{/privacy}","received_events_url":"https://api.github.com/users/HeartSaVioR/received_events","type":"User","site_admin":false},"created_at":"2019-08-27T09:19:50Z","updated_at":"2019-08-27T09:19:50Z","author_association":"CONTRIBUTOR","body":"> Regarding output mode, most of Beam runners (spark for ex) support discarding output in which element from different windows are independent and previous states are dropped.\r\n\r\nI'm not sure I understand it correctly. The point for Append mode is, output for specific key (key shouldn't be necessary to be windowed, but should include \"event time\" column) will be provided only once in any case, regardless of allowed lateness, no case of \"upsert\". If Beam doesn't close the window when watermark passes by (but still doesn't pass by allowed lateness) but triggers window and emits the output of window so far (so output could be emitted multiple times), it's not compatible with Spark's Append mode.\r\n\r\nstream-stream join should decide which \"event time\" should be taken even we change the way of storing event time, as there're two rows being joined. How Beam decides \"event time\" for new record from two records? In column based event time (current Spark), it should be hard to choose \"min\" or \"max\" of event time, as which column to pick as event time should be determined by query plan phase."}}
{"actor":{"display_login":"HeartSaVioR"}}