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 2022/10/25 15:22:43 UTC
[GitHub] [beam] tvalentyn opened a new pull request, #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
tvalentyn opened a new pull request, #23828:
URL: https://github.com/apache/beam/pull/23828
fix attempt for: #20528
------------------------
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`).
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] 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/get-started-contributing/#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
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)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006271297
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
another suggestion to modify accumulation_mode directly via `hot.windowing.accumulation_mode = AccumulationMode.DISCARDING` also doesn't work, maybe because this is assignment is then reverted while still at pipeline construction time, also it has side-effect on other pcolls (e.g. cold), and hot.windowing is not assignable directly
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006268348
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
@robertwb Regarding your suggestion offline to use _IdentityWindowFn. Did you mean to use it as follows:
```
precombined_hot = (
hot
| 'WindowIntoIdentityAndDiscarding' >> WindowInto(
_IdentityWindowFn(pcoll.windowing.windowfn.get_window_coder()),
accumulation_mode=AccumulationMode.DISCARDING)
| CombinePerKey(PreCombineFn())
| Map(StripNonce)
| 'WindowIntoOriginal' >> WindowInto(pcoll.windowing))
return ((cold, precombined_hot)
| Flatten()
| CombinePerKey(PostCombineFn()))
```
?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006272045
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
I think current PR is the best fix I have so far, alternative is to skip `WindowIntoDiscarding` only if `SlidingWindows` are detected instead of applying it whenever accumulation_mode=AccumulationMode.DISCARDING is detected.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on PR #23828:
URL: https://github.com/apache/beam/pull/23828#issuecomment-1292699085
retest this please
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Force discarding mode in with_fanout without rewindowing.
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1021950288
##########
sdks/python/apache_beam/transforms/combiners_test.py:
##########
@@ -563,6 +563,70 @@ def has_expected_values(actual):
assert_that(result, has_expected_values)
+ def test_combining_with_sliding_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(
+ ordered,
+ contains_exactly([0, 1, 2, 3], [0, 1, 2, 3, 5, 6, 7, 8],
+ [5, 6, 7, 8]))
+
+ result = (
+ p
+ | beam.Create([
+ window.TimestampedValue(0, Timestamp(seconds=1666707510)),
+ window.TimestampedValue(1, Timestamp(seconds=1666707511)),
+ window.TimestampedValue(2, Timestamp(seconds=1666707512)),
+ window.TimestampedValue(3, Timestamp(seconds=1666707513)),
+ window.TimestampedValue(5, Timestamp(seconds=1666707515)),
+ window.TimestampedValue(6, Timestamp(seconds=1666707516)),
+ window.TimestampedValue(7, Timestamp(seconds=1666707517)),
+ window.TimestampedValue(8, Timestamp(seconds=1666707518))
+ ])
+ | beam.WindowInto(window.SlidingWindows(10, 5))
+ | beam.CombineGlobally(beam.combiners.ToListCombineFn()).
+ without_defaults().with_fanout(2))
+ assert_that(result, has_expected_values)
+
+ def test_combining_with_session_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(ordered, contains_exactly([0, 1, 2, 3], [5, 6, 7, 8]))
+ # Different runners have different number of 15s, but there should
+ # be at least one 15.
+
+ result = (
+ p
+ | beam.Create([
+ window.TimestampedValue(0, Timestamp(seconds=1666707510)),
+ window.TimestampedValue(1, Timestamp(seconds=1666707511)),
+ window.TimestampedValue(2, Timestamp(seconds=1666707512)),
+ window.TimestampedValue(3, Timestamp(seconds=1666707513)),
+ window.TimestampedValue(5, Timestamp(seconds=1666707515)),
+ window.TimestampedValue(6, Timestamp(seconds=1666707516)),
+ window.TimestampedValue(7, Timestamp(seconds=1666707517)),
+ window.TimestampedValue(8, Timestamp(seconds=1666707518))
+ ])
+ | beam.WindowInto(window.Sessions(2))
+ | beam.CombineGlobally(beam.combiners.ToListCombineFn()).
+ without_defaults().with_fanout(2))
Review Comment:
also checked for 100 locally
##########
sdks/python/apache_beam/transforms/combiners_test.py:
##########
@@ -563,6 +563,70 @@ def has_expected_values(actual):
assert_that(result, has_expected_values)
+ def test_combining_with_sliding_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(
+ ordered,
+ contains_exactly([0, 1, 2, 3], [0, 1, 2, 3, 5, 6, 7, 8],
+ [5, 6, 7, 8]))
+
+ result = (
+ p
+ | beam.Create([
+ window.TimestampedValue(0, Timestamp(seconds=1666707510)),
+ window.TimestampedValue(1, Timestamp(seconds=1666707511)),
+ window.TimestampedValue(2, Timestamp(seconds=1666707512)),
+ window.TimestampedValue(3, Timestamp(seconds=1666707513)),
+ window.TimestampedValue(5, Timestamp(seconds=1666707515)),
+ window.TimestampedValue(6, Timestamp(seconds=1666707516)),
+ window.TimestampedValue(7, Timestamp(seconds=1666707517)),
+ window.TimestampedValue(8, Timestamp(seconds=1666707518))
+ ])
+ | beam.WindowInto(window.SlidingWindows(10, 5))
+ | beam.CombineGlobally(beam.combiners.ToListCombineFn()).
+ without_defaults().with_fanout(2))
+ assert_that(result, has_expected_values)
+
+ def test_combining_with_session_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(ordered, contains_exactly([0, 1, 2, 3], [5, 6, 7, 8]))
+ # Different runners have different number of 15s, but there should
+ # be at least one 15.
+
+ result = (
+ p
+ | beam.Create([
+ window.TimestampedValue(0, Timestamp(seconds=1666707510)),
+ window.TimestampedValue(1, Timestamp(seconds=1666707511)),
+ window.TimestampedValue(2, Timestamp(seconds=1666707512)),
+ window.TimestampedValue(3, Timestamp(seconds=1666707513)),
+ window.TimestampedValue(5, Timestamp(seconds=1666707515)),
+ window.TimestampedValue(6, Timestamp(seconds=1666707516)),
+ window.TimestampedValue(7, Timestamp(seconds=1666707517)),
+ window.TimestampedValue(8, Timestamp(seconds=1666707518))
+ ])
+ | beam.WindowInto(window.Sessions(2))
+ | beam.CombineGlobally(beam.combiners.ToListCombineFn()).
+ without_defaults().with_fanout(2))
Review Comment:
also checked with 100 locally
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006269124
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
I tried that, seems that using that transform makes tests fail, i can dig deeper to undestand the failure mode but wanted to check if I understood your intent correctly
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006272045
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
I think current PR is the best fix I have so far, alternative is to skip `WindowIntoDiscarding` explicitly if `SlidingWindows` are detected.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] commented on pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on PR #23828:
URL: https://github.com/apache/beam/pull/23828#issuecomment-1292731762
# [Codecov](https://codecov.io/gh/apache/beam/pull/23828?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#23828](https://codecov.io/gh/apache/beam/pull/23828?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (fe82dc0) into [master](https://codecov.io/gh/apache/beam/commit/a0f4e5e6adb9b53ce7d1cc9d318811fc6aba5b14?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (a0f4e5e) will **decrease** coverage by `0.06%`.
> The diff coverage is `100.00%`.
```diff
@@ Coverage Diff @@
## master #23828 +/- ##
==========================================
- Coverage 73.17% 73.10% -0.07%
==========================================
Files 720 728 +8
Lines 96044 96441 +397
==========================================
+ Hits 70278 70505 +227
- Misses 24454 24624 +170
Partials 1312 1312
```
| Flag | Coverage Δ | |
|---|---|---|
| python | `82.79% <100.00%> (-0.16%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/23828?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/python/apache\_beam/transforms/core.py](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHJhbnNmb3Jtcy9jb3JlLnB5) | `92.97% <100.00%> (+0.02%)` | :arrow_up: |
| [.../python/apache\_beam/testing/test\_stream\_service.py](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdGVzdGluZy90ZXN0X3N0cmVhbV9zZXJ2aWNlLnB5) | `88.09% <0.00%> (-4.77%)` | :arrow_down: |
| [sdks/python/apache\_beam/internal/gcp/auth.py](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW50ZXJuYWwvZ2NwL2F1dGgucHk=) | `79.16% <0.00%> (-3.70%)` | :arrow_down: |
| [...python/apache\_beam/runners/worker/worker\_status.py](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvd29ya2VyX3N0YXR1cy5weQ==) | `75.33% <0.00%> (-1.34%)` | :arrow_down: |
| [...che\_beam/runners/interactive/interactive\_runner.py](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9pbnRlcmFjdGl2ZV9ydW5uZXIucHk=) | `90.50% <0.00%> (-1.27%)` | :arrow_down: |
| [...teractive/testing/integration/notebook\_executor.py](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS90ZXN0aW5nL2ludGVncmF0aW9uL25vdGVib29rX2V4ZWN1dG9yLnB5) | `28.86% <0.00%> (-1.04%)` | :arrow_down: |
| [sdks/python/apache\_beam/transforms/combiners.py](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHJhbnNmb3Jtcy9jb21iaW5lcnMucHk=) | `93.05% <0.00%> (-0.39%)` | :arrow_down: |
| [sdks/python/apache\_beam/io/gcp/gcsio.py](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL2djc2lvLnB5) | `92.12% <0.00%> (-0.14%)` | :arrow_down: |
| [sdks/python/apache\_beam/ml/gcp/visionml.py](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvZ2NwL3Zpc2lvbm1sLnB5) | `92.59% <0.00%> (ø)` | |
| [sdks/python/apache\_beam/ml/gcp/visionml\_test\_it.py](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvZ2NwL3Zpc2lvbm1sX3Rlc3RfaXQucHk=) | `66.66% <0.00%> (ø)` | |
| ... and [19 more](https://codecov.io/gh/apache/beam/pull/23828/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1017163591
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
TODO: try to keep the same trigger, and/or other properties on the windowing mode.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006269124
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
I tried that, and it doesn't work. Seems like windows change, for example in the test I am adding, the elements become combined into 1 window, instead of 2. i can dig deeper to undestand the failure mode but wanted to check if I understood your intent correctly.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Force discarding mode in with_fanout without rewindowing.
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1021950026
##########
sdks/python/apache_beam/transforms/combiners_test.py:
##########
@@ -563,6 +563,70 @@ def has_expected_values(actual):
assert_that(result, has_expected_values)
+ def test_combining_with_sliding_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(
+ ordered,
+ contains_exactly([0, 1, 2, 3], [0, 1, 2, 3, 5, 6, 7, 8],
+ [5, 6, 7, 8]))
+
+ result = (
+ p
+ | beam.Create([
+ window.TimestampedValue(0, Timestamp(seconds=1666707510)),
+ window.TimestampedValue(1, Timestamp(seconds=1666707511)),
+ window.TimestampedValue(2, Timestamp(seconds=1666707512)),
+ window.TimestampedValue(3, Timestamp(seconds=1666707513)),
+ window.TimestampedValue(5, Timestamp(seconds=1666707515)),
+ window.TimestampedValue(6, Timestamp(seconds=1666707516)),
+ window.TimestampedValue(7, Timestamp(seconds=1666707517)),
+ window.TimestampedValue(8, Timestamp(seconds=1666707518))
+ ])
+ | beam.WindowInto(window.SlidingWindows(10, 5))
+ | beam.CombineGlobally(beam.combiners.ToListCombineFn()).
+ without_defaults().with_fanout(2))
+ assert_that(result, has_expected_values)
+
+ def test_combining_with_session_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(ordered, contains_exactly([0, 1, 2, 3], [5, 6, 7, 8]))
+ # Different runners have different number of 15s, but there should
+ # be at least one 15.
+
+ result = (
+ p
+ | beam.Create([
+ window.TimestampedValue(0, Timestamp(seconds=1666707510)),
+ window.TimestampedValue(1, Timestamp(seconds=1666707511)),
+ window.TimestampedValue(2, Timestamp(seconds=1666707512)),
+ window.TimestampedValue(3, Timestamp(seconds=1666707513)),
+ window.TimestampedValue(5, Timestamp(seconds=1666707515)),
+ window.TimestampedValue(6, Timestamp(seconds=1666707516)),
+ window.TimestampedValue(7, Timestamp(seconds=1666707517)),
+ window.TimestampedValue(8, Timestamp(seconds=1666707518))
+ ])
+ | beam.WindowInto(window.Sessions(2))
+ | beam.CombineGlobally(beam.combiners.ToListCombineFn()).
+ without_defaults().with_fanout(2))
Review Comment:
yes. increased the param slightly.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on pull request #23828: Force discarding mode in with_fanout without rewindowing.
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on PR #23828:
URL: https://github.com/apache/beam/pull/23828#issuecomment-1312373705
PTAL @robertwb
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006268348
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
@robertwb Regarding your suggestion offline to use _IdentityWindowFn. Did you mean to use instead:
```
precombined_hot = (
hot
| 'WindowIntoIdentityAndDiscarding' >> WindowInto(
_IdentityWindowFn(pcoll.windowing.windowfn.get_window_coder()), accumulation_mode=AccumulationMode.DISCARDING)
| CombinePerKey(PreCombineFn())
| Map(StripNonce)
| 'WindowIntoOriginal' >> WindowInto(pcoll.windowing))
return ((cold, precombined_hot)
| Flatten()
| CombinePerKey(PostCombineFn()))
```
?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Force discarding mode in with_fanout without rewindowing.
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1020678556
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
Updated per offline conversation.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006268348
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
@robertwb Regarding your suggestion offline to use _IdentityWindowFn. Did you mean to use instead:
```
precombined_hot = (
hot
| 'WindowIntoIdentityAndDiscarding' >> WindowInto(
_IdentityWindowFn(pcoll.windowing.windowfn.get_window_coder()),
accumulation_mode=AccumulationMode.DISCARDING)
| CombinePerKey(PreCombineFn())
| Map(StripNonce)
| 'WindowIntoOriginal' >> WindowInto(pcoll.windowing))
return ((cold, precombined_hot)
| Flatten()
| CombinePerKey(PostCombineFn()))
```
?
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
@robertwb Regarding your suggestion offline to use _IdentityWindowFn. Did you mean to use instead:
```
precombined_hot = (
hot
| 'WindowIntoIdentityAndDiscarding' >> WindowInto(
_IdentityWindowFn(pcoll.windowing.windowfn.get_window_coder()),
accumulation_mode=AccumulationMode.DISCARDING)
| CombinePerKey(PreCombineFn())
| Map(StripNonce)
| 'WindowIntoOriginal' >> WindowInto(pcoll.windowing))
return ((cold, precombined_hot)
| Flatten()
| CombinePerKey(PostCombineFn()))
```
?
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006269124
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
I tried that, and it doesn't work. Seems like windows change, for example in the test I am adding, the elements become combined into 1 window, instead of 3. i can dig deeper to undestand the failure mode but wanted to check if I understood your intent correctly.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on PR #23828:
URL: https://github.com/apache/beam/pull/23828#issuecomment-1290741794
R: @robertwb
cc: @chamikaramj
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] github-actions[bot] commented on pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #23828:
URL: https://github.com/apache/beam/pull/23828#issuecomment-1290834550
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006272045
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
I think current PR is the best fix I have so far, alternative is to skip `WindowIntoDiscarding` only if `SlidingWindows` are detected instead of applying `SlidingWindows` iff `accumulation_mode=AccumulationMode.DISCARDING` is detected.
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
I think current PR is the best fix I have so far, alternative is to skip `WindowIntoDiscarding` only if `SlidingWindows` are detected, instead of applying `SlidingWindows` iff `accumulation_mode=AccumulationMode.DISCARDING` is detected.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Adjust windowing strategy only for windows that explicitly set accumulating mode in with_fanout
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1006269124
##########
sdks/python/apache_beam/transforms/core.py:
##########
@@ -2779,14 +2779,20 @@ def StripNonce(nonce_key_value):
cold, hot = pcoll | ParDo(SplitHotCold()).with_outputs('hot', main='cold')
cold.element_type = typehints.Any # No multi-output type hints.
+ precombined_hot = hot
+ if pcoll.windowing.accumulation_mode == AccumulationMode.ACCUMULATING:
+ # Avoid double-counting that may happen with stacked accumulating mode.
+ precombined_hot |= 'WindowIntoDiscarding' >> WindowInto(
+ pcoll.windowing, accumulation_mode=AccumulationMode.DISCARDING)
+
precombined_hot = (
- hot
- # Avoid double counting that may happen with stacked accumulating mode.
- | 'WindowIntoDiscarding' >> WindowInto(
Review Comment:
I tried that, and it doesn't work, i can dig deeper to undestand the failure mode but wanted to check if I understood your intent correctly
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn merged pull request #23828: Force discarding mode in with_fanout without rewindowing.
Posted by GitBox <gi...@apache.org>.
tvalentyn merged PR #23828:
URL: https://github.com/apache/beam/pull/23828
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] robertwb commented on a diff in pull request #23828: Force discarding mode in with_fanout without rewindowing.
Posted by GitBox <gi...@apache.org>.
robertwb commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1021833690
##########
sdks/python/apache_beam/transforms/combiners_test.py:
##########
@@ -563,6 +563,70 @@ def has_expected_values(actual):
assert_that(result, has_expected_values)
+ def test_combining_with_sliding_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(
+ ordered,
+ contains_exactly([0, 1, 2, 3], [0, 1, 2, 3, 5, 6, 7, 8],
+ [5, 6, 7, 8]))
+
+ result = (
+ p
+ | beam.Create([
+ window.TimestampedValue(0, Timestamp(seconds=1666707510)),
+ window.TimestampedValue(1, Timestamp(seconds=1666707511)),
+ window.TimestampedValue(2, Timestamp(seconds=1666707512)),
+ window.TimestampedValue(3, Timestamp(seconds=1666707513)),
+ window.TimestampedValue(5, Timestamp(seconds=1666707515)),
+ window.TimestampedValue(6, Timestamp(seconds=1666707516)),
+ window.TimestampedValue(7, Timestamp(seconds=1666707517)),
+ window.TimestampedValue(8, Timestamp(seconds=1666707518))
+ ])
+ | beam.WindowInto(window.SlidingWindows(10, 5))
+ | beam.CombineGlobally(beam.combiners.ToListCombineFn()).
+ without_defaults().with_fanout(2))
+ assert_that(result, has_expected_values)
+
+ def test_combining_with_session_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(ordered, contains_exactly([0, 1, 2, 3], [5, 6, 7, 8]))
+ # Different runners have different number of 15s, but there should
+ # be at least one 15.
+
+ result = (
+ p
+ | beam.Create([
+ window.TimestampedValue(0, Timestamp(seconds=1666707510)),
+ window.TimestampedValue(1, Timestamp(seconds=1666707511)),
+ window.TimestampedValue(2, Timestamp(seconds=1666707512)),
+ window.TimestampedValue(3, Timestamp(seconds=1666707513)),
+ window.TimestampedValue(5, Timestamp(seconds=1666707515)),
+ window.TimestampedValue(6, Timestamp(seconds=1666707516)),
+ window.TimestampedValue(7, Timestamp(seconds=1666707517)),
+ window.TimestampedValue(8, Timestamp(seconds=1666707518))
+ ])
+ | beam.WindowInto(window.Sessions(2))
+ | beam.CombineGlobally(beam.combiners.ToListCombineFn()).
+ without_defaults().with_fanout(2))
Review Comment:
Does this work if we have with_fanout(100) as well?
##########
sdks/python/apache_beam/transforms/combiners_test.py:
##########
@@ -563,6 +563,70 @@ def has_expected_values(actual):
assert_that(result, has_expected_values)
+ def test_combining_with_sliding_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(
+ ordered,
+ contains_exactly([0, 1, 2, 3], [0, 1, 2, 3, 5, 6, 7, 8],
+ [5, 6, 7, 8]))
+
+ result = (
+ p
+ | beam.Create([
+ window.TimestampedValue(0, Timestamp(seconds=1666707510)),
+ window.TimestampedValue(1, Timestamp(seconds=1666707511)),
+ window.TimestampedValue(2, Timestamp(seconds=1666707512)),
+ window.TimestampedValue(3, Timestamp(seconds=1666707513)),
+ window.TimestampedValue(5, Timestamp(seconds=1666707515)),
+ window.TimestampedValue(6, Timestamp(seconds=1666707516)),
+ window.TimestampedValue(7, Timestamp(seconds=1666707517)),
+ window.TimestampedValue(8, Timestamp(seconds=1666707518))
+ ])
+ | beam.WindowInto(window.SlidingWindows(10, 5))
+ | beam.CombineGlobally(beam.combiners.ToListCombineFn()).
+ without_defaults().with_fanout(2))
+ assert_that(result, has_expected_values)
+
+ def test_combining_with_session_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(ordered, contains_exactly([0, 1, 2, 3], [5, 6, 7, 8]))
+ # Different runners have different number of 15s, but there should
Review Comment:
I'm not following this comment.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on a diff in pull request #23828: Force discarding mode in with_fanout without rewindowing.
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on code in PR #23828:
URL: https://github.com/apache/beam/pull/23828#discussion_r1021949854
##########
sdks/python/apache_beam/transforms/combiners_test.py:
##########
@@ -563,6 +563,70 @@ def has_expected_values(actual):
assert_that(result, has_expected_values)
+ def test_combining_with_sliding_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(
+ ordered,
+ contains_exactly([0, 1, 2, 3], [0, 1, 2, 3, 5, 6, 7, 8],
+ [5, 6, 7, 8]))
+
+ result = (
+ p
+ | beam.Create([
+ window.TimestampedValue(0, Timestamp(seconds=1666707510)),
+ window.TimestampedValue(1, Timestamp(seconds=1666707511)),
+ window.TimestampedValue(2, Timestamp(seconds=1666707512)),
+ window.TimestampedValue(3, Timestamp(seconds=1666707513)),
+ window.TimestampedValue(5, Timestamp(seconds=1666707515)),
+ window.TimestampedValue(6, Timestamp(seconds=1666707516)),
+ window.TimestampedValue(7, Timestamp(seconds=1666707517)),
+ window.TimestampedValue(8, Timestamp(seconds=1666707518))
+ ])
+ | beam.WindowInto(window.SlidingWindows(10, 5))
+ | beam.CombineGlobally(beam.combiners.ToListCombineFn()).
+ without_defaults().with_fanout(2))
+ assert_that(result, has_expected_values)
+
+ def test_combining_with_session_windows_and_fanout(self):
+ options = PipelineOptions()
+ options.view_as(StandardOptions).streaming = True
+ with TestPipeline(options=options) as p:
+
+ def has_expected_values(actual):
+ from hamcrest.core import assert_that as hamcrest_assert
+ from hamcrest.library.collection import contains_exactly
+ ordered = sorted(actual)
+
+ hamcrest_assert(ordered, contains_exactly([0, 1, 2, 3], [5, 6, 7, 8]))
+ # Different runners have different number of 15s, but there should
Review Comment:
copypaste leftover, removed.
--
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.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org