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/30 23:28:33 UTC
[GitHub] [beam] boyuanzz opened a new pull request #12430: [WIP][BEAM-10303] Scale progress with respect to windows observation.
boyuanzz opened a new pull request #12430:
URL: https://github.com/apache/beam/pull/12430
**Please** add a meaningful description for your change here
------------------------
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_Direct/lastCompletedBuild/) | --- | [](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/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_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] boyuanzz merged pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz merged pull request #12430:
URL: https://github.com/apache/beam/pull/12430
----------------------------------------------------------------
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 #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r470772348
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,12 +1048,36 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ return scaleProgress(
+ ((HasProgress) currentTracker).getProgress(),
+ currentWindowIterator.previousIndex(),
+ windowStopIndex);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindow != null) {
+ return scaleProgress(
+ Progress.from(elementCompleted, 1 - elementCompleted),
+ currentWindowIterator.previousIndex(),
+ windowStopIndex);
}
}
return null;
}
+ private Progress scaleProgress(Progress progress, int currentWindowIndex, int stopWindowIndex) {
Review comment:
nit: make the method static
----------------------------------------------------------------
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] boyuanzz commented on a change in pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r467266336
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
Are you saying that one transform can report the progress metric per element/window? For example, saying that the Truncate is on window1 and there are 3 windows in total, the downstream sdf progress is 0.5. Then the Truncate should report [(work_completed=1, work_remaining=0), (work_completed=1, work_remaining=0), (work_completed=0, work_remaining=1)] and the sdf should report [(work_completed=0.5, work_remaining=0.5)]? The the actual completed should be 1 + 1 * 0.5?
----------------------------------------------------------------
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 #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r465902065
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
Originally the idea was that we didn't want the SDK to have to perform these calculations and it is why each operator was going to report work_completed/work_remaining if it had them but it seems like accurate splitting by fraction needs to take it into account.
Using the graph to compute the progress shouldn't be any more/less difficult then the work that is being put into the SDK.
Is there still value in reporting the work_completed/work_remaining metrics then?
----------------------------------------------------------------
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] boyuanzz commented on a change in pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r467296625
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
I would prefer to let gRPC report element progress since it looks simpler, and also helps us to report less metrics. It's also more straight forward for each transform to compute element progress regarding to downstream progress.
----------------------------------------------------------------
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 #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r467178630
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,12 +1040,35 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ return scaleProgress(
+ ((HasProgress) currentTracker).getProgress(),
+ currentWindowIterator.previousIndex(),
+ currentElement.getWindows().size());
}
}
return null;
}
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
+ return scaleProgress(
+ Progress.from(elementCompleted, 1 - elementCompleted),
+ currentWindowIterator.previousIndex(),
+ currentElement.getWindows().size());
+ }
+ }
+ return null;
+ }
+
+ private Progress scaleProgress(Progress progress, int completedWindowIndex, int windowCount) {
Review comment:
For the future:
```suggestion
private Progress scaleProgress(Progress progress, int currentWindowIndex, int stopWindowIndex) {
```
Note that it is important that it isn't the number of completed windows but the current window index otherwise the math doesn't work out.
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,12 +1040,35 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ return scaleProgress(
+ ((HasProgress) currentTracker).getProgress(),
+ currentWindowIterator.previousIndex(),
+ currentElement.getWindows().size());
}
}
return null;
}
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
+ return scaleProgress(
+ Progress.from(elementCompleted, 1 - elementCompleted),
+ currentWindowIterator.previousIndex(),
+ currentElement.getWindows().size());
+ }
+ }
+ return null;
+ }
+
+ private Progress scaleProgress(Progress progress, int completedWindowIndex, int windowCount) {
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
Review comment:
`totalWork` -> `totalWorkPerWindow`
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,12 +1040,35 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ return scaleProgress(
+ ((HasProgress) currentTracker).getProgress(),
+ currentWindowIterator.previousIndex(),
+ currentElement.getWindows().size());
Review comment:
This should be the `stop index` since splitting limits the current window iterator so the progress will be wrong after a split happens.
----------------------------------------------------------------
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] boyuanzz commented on pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #12430:
URL: https://github.com/apache/beam/pull/12430#issuecomment-674342840
Run Java 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 #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r467182990
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
I did forget something, it turns out that `work_completed`/`work_remaining` are `lists` of progresses so you can report one for each `active` element so it turns out that the protocol didn't forget about this inprogress issue.
This could change our suggestion of how to report progress for window observing things or we can go with the idea of pushing it up to the gRPC read still.
----------------------------------------------------------------
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] boyuanzz commented on pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #12430:
URL: https://github.com/apache/beam/pull/12430#issuecomment-674332390
Run Java 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] boyuanzz commented on pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #12430:
URL: https://github.com/apache/beam/pull/12430#issuecomment-668168142
Currently `Truncate` didn't report progress via `addProgressRequestCallback`. It would result in incorrect progress when `Truncate` observing windows. If `Truncate` reports progress, it may require runner to change the logic of extracting element progress. For example, the runner should get progress from `Truncate` transform rather than `SDF/process`.
----------------------------------------------------------------
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 #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r464673014
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -515,6 +515,9 @@
&& Iterables.get(mainOutputConsumers, 0) instanceof HandlesSplits) {
mainInputConsumer =
new SplittableFnDataReceiver() {
+ private final HandlesSplits splitDelegate =
+ (HandlesSplits) Iterables.get(mainOutputConsumers, 0);
Review comment:
nit: here and below around line 558
```suggestion
(HandlesSplits) Iterables.getOnlyElement(mainOutputConsumers);
```
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
Should we register this with `addProgressRequestCallback` so we generate monitoring infos?
I'm not sure if truncate should be using the downstream progress as part of its calculation when reporting it as a monitoring info. I know that this differs from how we calculate the progress/split point for the SplittableFnDataReceiver since the singular fraction needs to take into account the downstream progress accurately.
I was always envisioning that work completed/work remaining for the monitoring infos always represented the local knowledge of work and didn't take into account any downstream/upstream knowledge. We can avoid this issue if we merge this logic into the `getProgress` method around line 533.
----------------------------------------------------------------
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] boyuanzz commented on a change in pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r465891961
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
Thanks for the formula. If we let the runner to do the computation, the runner needs to lookup the graph to figure out which progress is from downstream. I'm thinking about another option: we can add a progress signal for split specifically, which should be reported by the root node of the SDK graph. All computations you have mentioned can be done in the SDK side recursively. The runner side can decide whether to look into this signal by knowing whether there is an SDF(or SDFs) in the SDK graph.
----------------------------------------------------------------
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 #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r465304871
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
Thinking about this more, I do believe that the progress does need to be reported as a metric so a runner can choose a split fraction and also compute the amount of remaining work and/or completion time estimate. It looks like either:
1) Need to make work completed/remaining take into account downstream processing
OR
2) Need to add a metric that represents work in progress so that a runner can compute the amount of work being done (without this we can't figure out how much the work remaining downstream is relative to an upstream node).
I'm not a big fan of 1) since it means that this metric is intrinsically tied to the state of other transforms while in 2) we are adding something new.
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
Thinking about this more, I do believe that the progress does need to be reported as a metric so a runner can choose a split fraction and also compute the amount of remaining work and/or completion time estimate. It looks like either:
1) Need to make work completed/remaining take into account downstream processing
2) Need to add a metric that represents work in progress so that a runner can compute the amount of work being done (without this we can't figure out how much the work remaining downstream is relative to an upstream node).
I'm not a big fan of 1) since it means that this metric is intrinsically tied to the state of other transforms while in 2) we are adding something new.
----------------------------------------------------------------
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] boyuanzz commented on a change in pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r465381343
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
I think the combination of Truncate observing windows and Process consuming elements makes progress/split quite complicated. It seems like a special case of handling split/progress for a non-root SDF.
If we treat the Truncate only as a special case, we can go with option1 and let the runner to decide how to make the split decision. In this case, it's natural for Truncate to report the progress combing with the downstream progress.
For option2, I'm not sure how a runner computes the work remaining based on the new metric.
----------------------------------------------------------------
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 #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r465855425
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
Yes, truncate exposes some of the issues where a non root SDF makes things interesting.
For option2, if we assume that `inprogress` is included in `remaining` then:
```
total = completed + remaining
fraction_completed = (inprogress / remaining) * downstream_fraction_completed + completed / total
```
`downstream_fraction_completed` would be computed recursively and could also be computed effectively if there were multiple consumers with `(completed_conumers + downstream_fraction_completed) / total_num_consumers`
----------------------------------------------------------------
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] boyuanzz commented on a change in pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r467364484
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,12 +1040,35 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ return scaleProgress(
+ ((HasProgress) currentTracker).getProgress(),
+ currentWindowIterator.previousIndex(),
+ currentElement.getWindows().size());
Review comment:
Added `windowStopIndex`.
----------------------------------------------------------------
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] boyuanzz commented on a change in pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r465234180
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
I was hesitated to do so since I'm not sure whether the progress from Truncate is used anywhere unless it's used for runner side split decision. Then there is another question, should the progress from Truncate be used for split decision?
----------------------------------------------------------------
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 #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r467272382
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
Yes, assuming you meant to say that truncate is on window2.
----------------------------------------------------------------
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] boyuanzz commented on pull request #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on pull request #12430:
URL: https://github.com/apache/beam/pull/12430#issuecomment-674252988
Run Java 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 #12430: [BEAM-10303] Scale progress with respect to windows observation.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12430:
URL: https://github.com/apache/beam/pull/12430#discussion_r467182990
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1029,7 +1040,27 @@ public double getProgress() {
private Progress getProgress() {
synchronized (splitLock) {
if (currentTracker instanceof RestrictionTracker.HasProgress) {
- return ((HasProgress) currentTracker).getProgress();
+ Progress progress = ((HasProgress) currentTracker).getProgress();
+ double totalWork = progress.getWorkCompleted() + progress.getWorkRemaining();
+ double completed =
+ totalWork * currentWindowIterator.previousIndex() + progress.getWorkCompleted();
+ double remaining =
+ totalWork * (currentElement.getWindows().size() - currentWindowIterator.nextIndex())
+ + progress.getWorkRemaining();
+ return Progress.from(completed, remaining);
+ }
+ }
+ return null;
+ }
+
+ private Progress getProgressFromWindowObservingTruncate(double elementCompleted) {
+ synchronized (splitLock) {
+ if (currentWindowIterator != null) {
Review comment:
I did forget something, it turns out that `work_completed`/`work_remaining` are `lists` of progresses so you can report one for each `active` element so it turns out that the protocol didn't forget about this inprogress issue.
----------------------------------------------------------------
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