{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/21220","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/21220/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/21220/comments","events_url":"https://api.github.com/repos/apache/spark/issues/21220/events","html_url":"https://github.com/apache/spark/pull/21220","id":319731813,"number":21220,"title":"[SPARK-24157][SS] Enabled no-data batches in MicroBatchExecution for streaming aggregation and deduplication.","user":{"login":"tdas","id":663212,"avatar_url":"https://avatars1.githubusercontent.com/u/663212?v=4","gravatar_id":"","url":"https://api.github.com/users/tdas","html_url":"https://github.com/tdas","followers_url":"https://api.github.com/users/tdas/followers","following_url":"https://api.github.com/users/tdas/following{/other_user}","gists_url":"https://api.github.com/users/tdas/gists{/gist_id}","starred_url":"https://api.github.com/users/tdas/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tdas/subscriptions","organizations_url":"https://api.github.com/users/tdas/orgs","repos_url":"https://api.github.com/users/tdas/repos","events_url":"https://api.github.com/users/tdas/events{/privacy}","received_events_url":"https://api.github.com/users/tdas/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":6,"created_at":"2018-05-02T22:19:50Z","updated_at":"2018-05-04T08:31:46Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/21220","html_url":"https://github.com/apache/spark/pull/21220","diff_url":"https://github.com/apache/spark/pull/21220.diff","patch_url":"https://github.com/apache/spark/pull/21220.patch"},"body":"## What changes were proposed in this pull request?\r\n\r\nThis PR enables the MicroBatchExecution to run no-data batches if some SparkPlan requires running another batch to output results based on updated watermark / processing time. In this PR, I have enabled streaming aggregations and streaming deduplicates to automatically run addition batch even if new data is available. See https://issues.apache.org/jira/browse/SPARK-24156 for more context.\r\n\r\nMajor changes/refactoring done in this PR.\r\n- Refactoring MicroBatchExecution - A major point of confusion in MicroBatchExecution control flow was always (at least to me) was that `populateStartOffsets` internally called `constructNextBatch` which was not obvious from just the name \"populateStartOffsets\" and made the control flow from the main trigger execution loop very confusing (main loop in `runActivatedStream` called `constructNextBatch` but only if `populateStartOffsets` hadn't already called it). Instead, the refactoring makes it cleaner.\r\n    - `populateStartOffsets` only the updates `availableOffsets` and `committedOffsets`. Does not call `constructNextBatch`.\r\n    - Main loop in `runActivatedStream` calls `constructNextBatch` which returns true or false reflecting whether the next batch is ready for executing. This method is now idempotent; if a batch has already been constructed, then it will always return true until the batch has been executed.\r\n    - If next batch is ready then we call `runBatch` or sleep. \r\n    - That's it.\r\n\r\n- Refactoring watermark management logic - This has been refactored out from `MicroBatchExecution` in a separate class to simplify `MicroBatchExecution`.\r\n\r\n- New method `shouldRunAnotherBatch` in `IncrementalExecution` - This returns true if there is any stateful operation in the last execution plan that requires another batch for state cleanup, etc. This is used to decide whether to construct a batch or not in `constructNextBatch`.\r\n\r\n- Changes to stream testing framework - Many tests used CheckLastBatch to validate answers. This assumed that there will be no more batches after the last set of input has been processed, so the last batch is the one that has output corresponding to the last input. This is not true anymore. To account for that, I made two changes.\r\n    - `CheckNewAnswer` is a new test action that verifies the new rows generated since the last time the answer was checked by `CheckAnswer`, `CheckNewAnswer` or `CheckLastBatch`. This is agnostic to how many batches occurred between the last check and now. To do make this easier, I added a common trait between MemorySink and MemorySinkV2 to abstract out some common methods.\r\n    - `assertNumStateRows` has been updated in the same way to be agnostic to batches while checking what the total rows and how many state rows were updated (sums up updates since the last check).\r\n\r\n\r\n\r\n## How was this patch tested?\r\n- Changes made to existing tests - Tests have been changed in one of the following patterns.\r\n    - Tests where the last input was given again to force another batch to be executed and state cleaned up / output generated, they were simplified by removing the extra input.\r\n    - Tests using aggregation+watermark where CheckLastBatch were replaced with CheckNewAnswer to make them batch agnostic.\r\n- New tests added to check whether the flag works for streaming aggregation and deduplication\r\n\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/386536973","html_url":"https://github.com/apache/spark/pull/21220#issuecomment-386536973","issue_url":"https://api.github.com/repos/apache/spark/issues/21220","id":386536973,"user":{"login":"tdas","id":663212,"avatar_url":"https://avatars1.githubusercontent.com/u/663212?v=4","gravatar_id":"","url":"https://api.github.com/users/tdas","html_url":"https://github.com/tdas","followers_url":"https://api.github.com/users/tdas/followers","following_url":"https://api.github.com/users/tdas/following{/other_user}","gists_url":"https://api.github.com/users/tdas/gists{/gist_id}","starred_url":"https://api.github.com/users/tdas/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/tdas/subscriptions","organizations_url":"https://api.github.com/users/tdas/orgs","repos_url":"https://api.github.com/users/tdas/repos","events_url":"https://api.github.com/users/tdas/events{/privacy}","received_events_url":"https://api.github.com/users/tdas/received_events","type":"User","site_admin":false},"created_at":"2018-05-04T08:31:46Z","updated_at":"2018-05-04T08:31:46Z","author_association":"CONTRIBUTOR","body":"@brkyvz Answers to your questions.\r\n\r\n1. We have already fixed those emptyDF optimizations. The optimization only quick in `df.isStreaming = false`, and emptyDFs generated by sources should be have isStreaming = true. That's for v1 sources. For v2 sources, the engine already takes care of that by making sure that `StreamingDataSourceV2Relation.isStreaming` is true where `StreamingDataSourceV2Relation` is the logical plan leaf inserted into the micro-batch logical plan irrespective of empty or not.\r\n\r\n2. I will do both type of timeouts in flatMapGroupsWithState in later PR.\r\n"}}
{"actor":{"display_login":"tdas"}}