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/29 08:38:54 UTC
[GitHub] [beam] tvalentyn opened a new pull request #13220: Optimizes extract_output for 1 element accumulator case.
tvalentyn opened a new pull request #13220:
URL: https://github.com/apache/beam/pull/13220
Simple optimization for {CallableWrapperCombineFn, NoSideInputsCallableWrapperCombineFn}.extract_output that takes advantage of the corresponding (optional) compact behavior.
(Prepared by @katsiapis).
------------------------
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] yifanmai commented on pull request #13220: Optimizes extract_output for 1 element accumulator case.
Posted by GitBox <gi...@apache.org>.
yifanmai commented on pull request #13220:
URL: https://github.com/apache/beam/pull/13220#issuecomment-754810658
`f(a, b, ...) = 2 * sum(a, b, ...)` is not associative:
```
f(f(a, b), c) = 4a + 4b + 2c
f(a, f(b, c)) = 2a + 4b + 4c
f(f(a, b), c) != f(a, f(b, c))
```
A better example would be sum modulo n i.e. `f(a, b, ...) = sum(a, b, ...) % n`
----------------------------------------------------------------
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] tvalentyn commented on pull request #13220: Optimizes extract_output for 1 element accumulator case.
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on pull request #13220:
URL: https://github.com/apache/beam/pull/13220#issuecomment-750674335
We need to revert this change.
Beam Combiners are associative and commutative reduction operators. It is reasonable to assume that the input `fn` to [`CombineFn.from_callable()`](https://github.com/apache/beam/blob/fbeb28e7d974c79a8edf00ac80c72fe0ef35f293/sdks/python/apache_beam/transforms/core.py#L980) must be an associative and commutative reduction. Being idempotent on single-element inputs not a requirement. There can be potentially useful combiners that are not idempotent, for example a combiner that checks whether a particular element is in a PCollection, generated by `CombineFn.from_callable(lambda input_list: 'some_element' in input_list)`.
Beam programming guide[1] also mentions this:
> When you apply a Combine transform, you must provide the function that contains the logic for combining the elements or values. The combining function should be commutative and associative, as the function is not necessarily invoked exactly once on all values with a given key.
The implementation of `CallableWrapperCombineFn`/`NoSideInputsCallableWrapperCombineFn` should guarantee that if the input fn is an associative and commutative reduction, the resulting combiner produced by the result of wrapping the fn will also be an associative and commutative reduction, producing the same behavior.
With this change, we no longer have this guarantee. Calls to [CallableWrapperCombineFn.add_input()](https://github.com/apache/beam/blob/fbeb28e7d974c79a8edf00ac80c72fe0ef35f293/sdks/python/apache_beam/transforms/core.py#L1056) append an element to an accumulator, but if there is only one accumulator with only one element, we assume it is already a combined value and don't call `self._fn()` (the intent of the optimization in this change). For non-idempotent reducers, this is not the case.
----------------------------------------------------------------
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] tvalentyn commented on pull request #13220: Optimizes extract_output for 1 element accumulator case.
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on pull request #13220:
URL: https://github.com/apache/beam/pull/13220#issuecomment-718482152
R: @yifanmai
cc: @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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tvalentyn commented on pull request #13220: Optimizes extract_output for 1 element accumulator case.
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on pull request #13220:
URL: https://github.com/apache/beam/pull/13220#issuecomment-750675355
thanks, @y1chi who flagged it.
----------------------------------------------------------------
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] tvalentyn merged pull request #13220: Optimizes extract_output for 1 element accumulator case.
Posted by GitBox <gi...@apache.org>.
tvalentyn merged pull request #13220:
URL: https://github.com/apache/beam/pull/13220
----------------------------------------------------------------
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] tvalentyn commented on pull request #13220: Optimizes extract_output for 1 element accumulator case.
Posted by GitBox <gi...@apache.org>.
tvalentyn commented on pull request #13220:
URL: https://github.com/apache/beam/pull/13220#issuecomment-750926773
`CombineFn.from_callable(lambda input_list: 'some_element' in input_list)` would not work as a combiner since the fn is not associative/commutative. `CombineFn.from_callable(lambda input_list: 2*sum(input_list))` is a better example.
----------------------------------------------------------------
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] tvalentyn edited a comment on pull request #13220: Optimizes extract_output for 1 element accumulator case.
Posted by GitBox <gi...@apache.org>.
tvalentyn edited a comment on pull request #13220:
URL: https://github.com/apache/beam/pull/13220#issuecomment-750674335
We need to revert this change.
Beam Combiners are associative and commutative reduction operators. It is reasonable to assume that the input `fn` to [`CombineFn.from_callable()`](https://github.com/apache/beam/blob/fbeb28e7d974c79a8edf00ac80c72fe0ef35f293/sdks/python/apache_beam/transforms/core.py#L980) must be an associative and commutative reduction. Being idempotent on single-element inputs not a requirement. There can be combiners that are not idempotent, for example a combiner for example one generated by `CombineFn.from_callable(lambda input_list: 2*sum(input_list))`.
Beam programming guide[1] also mentions this:
> When you apply a Combine transform, you must provide the function that contains the logic for combining the elements or values. The combining function should be commutative and associative, as the function is not necessarily invoked exactly once on all values with a given key.
The implementation of `CallableWrapperCombineFn`/`NoSideInputsCallableWrapperCombineFn` should guarantee that if the input fn is an associative and commutative reduction, the resulting combiner produced by the result of wrapping the fn will also be an associative and commutative reduction, producing the same behavior.
With this change, we no longer have this guarantee. Calls to [CallableWrapperCombineFn.add_input()](https://github.com/apache/beam/blob/fbeb28e7d974c79a8edf00ac80c72fe0ef35f293/sdks/python/apache_beam/transforms/core.py#L1056) append an element to an accumulator, but if there is only one accumulator with only one element, we assume it is already a combined value and don't call `self._fn()` (the intent of the optimization in this change). For non-idempotent reducers, this is not the case.
Edit: updated the example fn.
----------------------------------------------------------------
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