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/10/09 19:05:23 UTC
[GitHub] [beam] lukecwik opened a new pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
lukecwik opened a new pull request #13061:
URL: https://github.com/apache/beam/pull/13061
Accumulators can be mutated during merging by the combine fn so we must ensure that we use a unique instance of the accumulator per window.
------------------------
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://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](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_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_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 | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
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/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_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.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
----------------------------------------------------------------
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 #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-706354261
R: @iemejia
----------------------------------------------------------------
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 #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-706490005
Run Spark StructuredStreaming ValidatesRunner
----------------------------------------------------------------
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] iemejia commented on pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-723113493
@echauchot The VR tests were breaking on this (I don't know why, maybe the tests were improved). That's the reason why Luke did this PR, it was needed at least for correctness. You can reproduce this by reverting this PR and running the tests:
```
git revert 6264b47afd51d33d95d6c04a2106b4208a89ca41
./gradlew :runners:spark:validatesStructuredStreamingRunnerBatch
```
produces
```
org.apache.beam.sdk.transforms.CombineTest$WindowingTests > testSlidingWindowsCombine FAILED
java.lang.AssertionError at CombineTest.java:1156
Caused by: java.lang.AssertionError at MatcherAssert.java:18
```
Something odd I noticed is that if you run the single test instance it passes so I am not sure if there is some interleaving issue with other tests.
The VR suite of the Structured Streaming Runner has been broken since September 10 also because of BEAM-11023
http://104.154.241.245/d/8N6LVeCmk/post-commits-status-dashboard?refresh=30s&orgId=1
----------------------------------------------------------------
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] echauchot edited a comment on pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
echauchot edited a comment on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-722451553
@lukecwik I don't see why this change is necessary because of 2 reasons:
1. all the validates runner tests including multiple window (eg. sliding windows) already passed.
2. when I wrote this code, I already took some safety mesures about the modification of the (first only) accumulator during the `combineFn.mergeAccumulator` by creating a new first accumulator for each merged window see initial code below:
// merge the accumulators for each mergedWindow
...
for (Map.Entry<W, List<Tuple2<AccumT, Instant>>> entry :
mergedWindowToAccumulators.entrySet()) {
...
// we need to create the first accumulator because combineFn.mergerAccumulators can modify the
// first accumulator
AccumT first = combineFn.createAccumulator();
Iterable<AccumT> accumulatorsToMerge =
Iterables.concat(
Collections.singleton(first),
accumsAndInstantsForMergedWindow.stream()
.map(x -> x._1())
.collect(Collectors.toList()));
...
combineFn.mergeAccumulators(accumulatorsToMerge),
...
}
----------------------------------------------------------------
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 #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-707236966
I would prefer taking the fix and then further optimizing for performance as the implementation I suggested only duplicates when a value is in multiple windows which is uncommon in practice.
----------------------------------------------------------------
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] echauchot commented on pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
echauchot commented on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-722451553
@lukecwik I don't see why this change is necessary because of 2 reasons:
1. all the validates runner tests including multiple window (eg. sliding windows) already passed.
2. when I wrote this code, I already took some safety mesures about the modification of the (first only) accumulator during the `combineFn.mergeAccumulator` by creating a new first accumulator for each merged window see initial code below:
` // merge the accumulators for each mergedWindow
...
for (Map.Entry<W, List<Tuple2<AccumT, Instant>>> entry :
mergedWindowToAccumulators.entrySet()) {
...
// we need to create the first accumulator because combineFn.mergerAccumulators can modify the
// first accumulator
AccumT first = combineFn.createAccumulator();
Iterable<AccumT> accumulatorsToMerge =
Iterables.concat(
Collections.singleton(first),
accumsAndInstantsForMergedWindow.stream()
.map(x -> x._1())
.collect(Collectors.toList()));
...
combineFn.mergeAccumulators(accumulatorsToMerge),
...
}
`
----------------------------------------------------------------
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 #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-706492265
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] echauchot edited a comment on pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
echauchot edited a comment on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-722451553
@lukecwik I don't see why this change is necessary because of 2 reasons:
1. all the validates runner tests including multiple window (eg. sliding windows) already passed.
2. when I wrote this code, I already took some safety mesures about the modification of the (first only) accumulator during the `combineFn.mergeAccumulator` by creating a new first accumulator for each merged window see initial code below:
```
// merge the accumulators for each mergedWindow
...
for (Map.Entry<W, List<Tuple2<AccumT, Instant>>> entry :
mergedWindowToAccumulators.entrySet()) {
...
// we need to create the first accumulator because combineFn.mergerAccumulators can modify the
// first accumulator
AccumT first = combineFn.createAccumulator();
Iterable<AccumT> accumulatorsToMerge =
Iterables.concat(
Collections.singleton(first),
accumsAndInstantsForMergedWindow.stream()
.map(x -> x._1())
.collect(Collectors.toList()));
...
combineFn.mergeAccumulators(accumulatorsToMerge),
...
}
```
----------------------------------------------------------------
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] iemejia edited a comment on pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
iemejia edited a comment on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-723113493
@echauchot The VR tests were breaking on this (I don't know why, maybe the tests were improved). That's the reason why Luke did this PR, it was needed at least for correctness. You can reproduce this by reverting this PR and running the tests:
```
git revert 6264b47afd51d33d95d6c04a2106b4208a89ca41
./gradlew :runners:spark:validatesStructuredStreamingRunnerBatch
```
produces
```
org.apache.beam.sdk.transforms.CombineTest$WindowingTests > testSlidingWindowsCombine FAILED
java.lang.AssertionError at CombineTest.java:1156
Caused by: java.lang.AssertionError at MatcherAssert.java:18
```
Something odd I noticed is that if you run the single test instance it passes so I am not sure if there is some interleaving issue with other tests.
The VR suite of the Structured Streaming Runner has been broken since September 10 also because of this issue and BEAM-11023 too.
http://104.154.241.245/d/8N6LVeCmk/post-commits-status-dashboard?refresh=30s&orgId=1
----------------------------------------------------------------
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] iemejia commented on pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-707325762
Good point, I suppose if @echauchot has a suggestion or a better way to do this we can improve it in the future, at least this fixes the breakage on tests and it produces correct results.
----------------------------------------------------------------
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] echauchot edited a comment on pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
echauchot edited a comment on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-722451553
@lukecwik I don't see why this change is necessary because of 2 reasons:
1. all the validates runner tests including multiple window (eg. sliding windows) already passed.
2. when I wrote this code, I already took some safety mesures about the modification of the (first only) accumulator during the `combineFn.mergeAccumulator` by creating a new first accumulator for each merged window see initial code below and the comment in the code:
```
// merge the accumulators for each mergedWindow
...
for (Map.Entry<W, List<Tuple2<AccumT, Instant>>> entry :
mergedWindowToAccumulators.entrySet()) {
...
// we need to create the first accumulator because combineFn.mergerAccumulators can modify the
// first accumulator
AccumT first = combineFn.createAccumulator();
Iterable<AccumT> accumulatorsToMerge =
Iterables.concat(
Collections.singleton(first),
accumsAndInstantsForMergedWindow.stream()
.map(x -> x._1())
.collect(Collectors.toList()));
...
combineFn.mergeAccumulators(accumulatorsToMerge),
...
}
```
----------------------------------------------------------------
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] echauchot commented on pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
echauchot commented on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-723128931
thanks @iemejia for the context. Strange `org.apache.beam.sdk.transforms.CombineTest$WindowingTests > testSlidingWindowsCombine ` was passing. So I would prefer to figure out what was changed in the between in the Combine translation. https://issues.apache.org/jira/browse/BEAM-11023 refers to groupByKey so it is unrelated. But I just checked it was passing when the runner was merged to master. I guess we need to dig into the history of commits rather
----------------------------------------------------------------
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 #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-706560252
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] iemejia merged pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
iemejia merged pull request #13061:
URL: https://github.com/apache/beam/pull/13061
----------------------------------------------------------------
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] iemejia commented on pull request #13061: [BEAM-11050] Duplicate accumulator if it appears in multiple windows.
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #13061:
URL: https://github.com/apache/beam/pull/13061#issuecomment-706571617
This LGTM but I prefer that @echauchot takes a look before merging because he has been optimizing this code for a while so better to make him aware of the issue and the minor performance hit of the extra encoding needed.
----------------------------------------------------------------
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