You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/07/08 02:54:28 UTC
[GitHub] [beam] lukecwik opened a new pull request #12192: [WIP][BEAM-10420] Migrate SDF logic into PerWindowInvoker
lukecwik opened a new pull request #12192:
URL: https://github.com/apache/beam/pull/12192
To be able to have the windowing optimization stay within the PerWindowInvoker, I needed to have it control the creation of the watermark estimator and the restriction tracker thus I supply the restrication and watermark estimator state.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | --- | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] chamikaramj commented on a change in pull request #12192: [BEAM-10420] Move restriction/watermark estimator state into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
chamikaramj commented on a change in pull request #12192:
URL: https://github.com/apache/beam/pull/12192#discussion_r452675861
##########
File path: sdks/python/apache_beam/runners/common.py
##########
@@ -661,8 +661,8 @@ def __init__(self, placeholder):
def invoke_process(self,
windowed_value, # type: WindowedValue
- restriction_tracker=None, # type: Optional[RestrictionTracker]
- watermark_estimator=None, # type: Optional[WatermarkEstimator]
+ restriction=None,
Review comment:
New type is just 'object' ?
##########
File path: sdks/python/apache_beam/runners/common.py
##########
@@ -661,8 +661,8 @@ def __init__(self, placeholder):
def invoke_process(self,
windowed_value, # type: WindowedValue
- restriction_tracker=None, # type: Optional[RestrictionTracker]
- watermark_estimator=None, # type: Optional[WatermarkEstimator]
+ restriction=None,
Review comment:
Mention when this can be None.
##########
File path: sdks/python/apache_beam/runners/common.py
##########
@@ -661,8 +661,8 @@ def __init__(self, placeholder):
def invoke_process(self,
windowed_value, # type: WindowedValue
- restriction_tracker=None, # type: Optional[RestrictionTracker]
- watermark_estimator=None, # type: Optional[WatermarkEstimator]
+ restriction=None,
+ watermark_estimator_state=None,
Review comment:
This is a tuple ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #12192: [BEAM-10420] Move restriction/watermark estimator state into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12192:
URL: https://github.com/apache/beam/pull/12192#discussion_r452922911
##########
File path: sdks/python/apache_beam/runners/common.py
##########
@@ -661,8 +661,8 @@ def __init__(self, placeholder):
def invoke_process(self,
windowed_value, # type: WindowedValue
- restriction_tracker=None, # type: Optional[RestrictionTracker]
- watermark_estimator=None, # type: Optional[WatermarkEstimator]
+ restriction=None,
+ watermark_estimator_state=None,
Review comment:
Its a user chosen value like the restriction. They can store anything they want to restore the state for the estimator. For system provided ones its typically the last reported timestamp.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #12192: [BEAM-10420] Move restriction/watermark estimator state into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12192:
URL: https://github.com/apache/beam/pull/12192#discussion_r452949109
##########
File path: sdks/python/apache_beam/runners/common.py
##########
@@ -661,8 +661,8 @@ def __init__(self, placeholder):
def invoke_process(self,
windowed_value, # type: WindowedValue
- restriction_tracker=None, # type: Optional[RestrictionTracker]
- watermark_estimator=None, # type: Optional[WatermarkEstimator]
+ restriction=None,
Review comment:
Done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #12192: [WIP][BEAM-10420] Migrate SDF logic into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12192:
URL: https://github.com/apache/beam/pull/12192#discussion_r451243088
##########
File path: sdks/python/apache_beam/runners/direct/sdf_direct_runner.py
##########
@@ -464,19 +461,15 @@ def initiate_checkpoint():
with self._checkpoint_lock:
if checkpoint_state.checkpointed:
return
- checkpoint_state.residual_restriction = tracker.checkpoint()
- checkpoint_state.checkpointed = object()
+ checkpoint_state.checkpointed = object()
Review comment:
This is a bug fix, we should be setting checkpoint_state.checkpointed while holding the lock.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik edited a comment on pull request #12192: [WIP][BEAM-10420] Migrate SDF logic into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik edited a comment on pull request #12192:
URL: https://github.com/apache/beam/pull/12192#issuecomment-655239038
R: @robertwb @chamikaramj
Progress reporting needs to be able to see which window out of how many are being processed. Splitting needs to be able to see which window is being processed to compute the primary and residual correctly.
Alternatively, I could move the window observing logic outside of the DoFnInvoker and then all the DoFnRunners would own this logic (this does seem like it would lead to more code duplication).
WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik closed pull request #12192: [WIP][BEAM-10420] Migrate SDF logic into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik closed pull request #12192:
URL: https://github.com/apache/beam/pull/12192
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #12192: [WIP][BEAM-10420] Migrate SDF logic into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #12192:
URL: https://github.com/apache/beam/pull/12192#issuecomment-655239038
R: @robertwb @chamikaramj
Style-wise does it make sense to use the `kwargs` to pass these through or should I continue to make them explicit like the `restriction_tracker` and `watermark_estimator` that was provided before.
Alternatively, I could move the window observing logic outside of the DoFnInvoker and then all the DoFnRunners would own this logic.
WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik edited a comment on pull request #12192: [WIP][BEAM-10420] Migrate SDF logic into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik edited a comment on pull request #12192:
URL: https://github.com/apache/beam/pull/12192#issuecomment-655239038
R: @robertwb @chamikaramj
Style-wise does it make sense to use the `kwargs` to pass these through or should I continue to make them explicit like the `restriction_tracker` and `watermark_estimator` that was provided before.
Alternatively, I could move the window observing logic outside of the DoFnInvoker and then all the DoFnRunners would own this logic (this does seem like it would lead to more code duplication).
WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #12192: [BEAM-10420] Move restriction/watermark estimator state into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12192:
URL: https://github.com/apache/beam/pull/12192#discussion_r452921776
##########
File path: sdks/python/apache_beam/runners/common.py
##########
@@ -661,8 +661,8 @@ def __init__(self, placeholder):
def invoke_process(self,
windowed_value, # type: WindowedValue
- restriction_tracker=None, # type: Optional[RestrictionTracker]
- watermark_estimator=None, # type: Optional[WatermarkEstimator]
+ restriction=None,
Review comment:
Yeah, since the user can choose anything they want for the restriction and watermark estimator state type.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #12192: [WIP][BEAM-10420] Migrate SDF logic into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #12192:
URL: https://github.com/apache/beam/pull/12192#issuecomment-655709449
Run Python PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #12192: [BEAM-10420] Move restriction/watermark estimator state into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #12192:
URL: https://github.com/apache/beam/pull/12192#issuecomment-656786730
Run Python PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #12192: [WIP][BEAM-10420] Migrate SDF logic into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12192:
URL: https://github.com/apache/beam/pull/12192#discussion_r451242979
##########
File path: sdks/python/apache_beam/runners/direct/sdf_direct_runner.py
##########
@@ -507,7 +500,6 @@ def initiate_checkpoint():
if self._max_num_outputs and output_count >= self._max_num_outputs:
initiate_checkpoint()
- tracker.check_done()
Review comment:
This is handled within the PerWindowInvoker already.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik merged pull request #12192: [BEAM-10420] Move restriction/watermark estimator state into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik merged pull request #12192:
URL: https://github.com/apache/beam/pull/12192
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik edited a comment on pull request #12192: [WIP][BEAM-10420] Migrate SDF logic into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik edited a comment on pull request #12192:
URL: https://github.com/apache/beam/pull/12192#issuecomment-655239038
R: @robertwb @chamikaramj
Progress reporting needs to be able to see which window out of how many are being processed. Splitting needs to be able to see which window is being processed to compute the primary and residual correctly.
Style-wise does it make sense to use the `kwargs` to pass these through or should I continue to make them explicit like the `restriction_tracker` and `watermark_estimator` that was provided before.
Alternatively, I could move the window observing logic outside of the DoFnInvoker and then all the DoFnRunners would own this logic (this does seem like it would lead to more code duplication).
WDYT?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #12192: [BEAM-10420] Move restriction/watermark estimator state into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12192:
URL: https://github.com/apache/beam/pull/12192#discussion_r452925354
##########
File path: sdks/python/apache_beam/runners/common.py
##########
@@ -661,8 +661,8 @@ def __init__(self, placeholder):
def invoke_process(self,
windowed_value, # type: WindowedValue
- restriction_tracker=None, # type: Optional[RestrictionTracker]
- watermark_estimator=None, # type: Optional[WatermarkEstimator]
+ restriction=None,
Review comment:
Done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #12192: [BEAM-10420] Move restriction/watermark estimator state into PerWindowInvoker
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #12192:
URL: https://github.com/apache/beam/pull/12192#issuecomment-656771326
Run Portable_Python PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org