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/14 16:39:09 UTC
[GitHub] [beam] davidcavazos opened a new pull request #12252: [BEAM-7390] Add combineglobally code snippets
davidcavazos opened a new pull request #12252:
URL: https://github.com/apache/beam/pull/12252
Add `CombineGlobally` code snippets.
R: @aaltay
R: @rosetn
------------------------
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/) | [](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] davidcavazos commented on pull request #12252: [BEAM-7390] Add combineglobally code snippets
Posted by GitBox <gi...@apache.org>.
davidcavazos commented on pull request #12252:
URL: https://github.com/apache/beam/pull/12252#issuecomment-658939252
I also noticed that the output tags were a little misleading now since we are showing the stdout instead of an output PCollection
----------------------------------------------------------------
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] aaltay merged pull request #12252: [BEAM-7390] Add combineglobally code snippets
Posted by GitBox <gi...@apache.org>.
aaltay merged pull request #12252:
URL: https://github.com/apache/beam/pull/12252
----------------------------------------------------------------
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] aaltay commented on a change in pull request #12252: [BEAM-7390] Add combineglobally code snippets
Posted by GitBox <gi...@apache.org>.
aaltay commented on a change in pull request #12252:
URL: https://github.com/apache/beam/pull/12252#discussion_r454689609
##########
File path: website/www/site/content/en/documentation/transforms/python/aggregation/combineglobally.md
##########
@@ -14,29 +14,197 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
+
# CombineGlobally
-<table align="left">
- <a target="_blank" class="button"
- href="https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineGlobally">
- <img src="https://beam.apache.org/images/logos/sdks/python.png" width="20px" height="20px"
- alt="Pydoc" />
- Pydoc
- </a>
-</table>
-<br><br>
+{{< localstorage language language-py >}}
+{{< button-pydoc path="apache_beam.transforms.core" class="CombineGlobally" >}}
Combines all elements in a collection.
See more information in the [Beam Programming Guide](/documentation/programming-guide/#combine).
## Examples
-See [BEAM-7390](https://issues.apache.org/jira/browse/BEAM-7390) for updates.
-## Related transforms
+In the following examples, we create a pipeline with a `PCollection` of produce.
+Then, we apply `CombineGlobally` in multiple ways to combine all the elements in the `PCollection`.
+
+`CombineGlobally` accepts a function that takes a list of elements as an input, and combines them to return a single element.
+
+### Example 1: Combining with a function
+
+We define a function `get_common_items` which takes a list of sets as an input, and calculates the intersection (common items) of those sets.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_function >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 2: Combining with a lambda function
+
+We can also use lambda functions to simplify **Example 1**.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_lambda >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 3: Combining with multiple arguments
+
+You can pass functions with multiple arguments to `CombineGlobally`.
+They are passed as additional positional arguments or keyword arguments to the function.
+
+In this example, the lambda function takes `sets` and `exclude` as arguments.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_multiple_arguments >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items_with_exceptions >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 4: Combining with side inputs as singletons
+
+If the `PCollection` has a single value, such as the average from another computation,
+passing the `PCollection` as a *singleton* accesses that value.
+
+In this example, we pass a `PCollection` the value `'🥕'` as a singleton.
+We then use that value to exclude specific items.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_side_inputs_singleton >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items_with_exceptions >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 5: Combining with side inputs as iterators
+
+If the `PCollection` has multiple values, pass the `PCollection` as an *iterator*.
+This accesses elements lazily as they are needed,
+so it is possible to iterate over large `PCollection`s that won't fit into memory.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_side_inputs_iter >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items_with_exceptions >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+> **Note**: You can pass the `PCollection` as a *list* with `beam.pvalue.AsList(pcollection)`,
+> but this requires that all the elements fit into memory.
+
+### Example 6: Combining with side inputs as dictionaries
+
+If a `PCollection` is small enough to fit into memory, then that `PCollection` can be passed as a *dictionary*.
+Each element must be a `(key, value)` pair.
+Note that all the elements of the `PCollection` must fit into memory for this.
+If the `PCollection` won't fit into memory, use `beam.pvalue.AsIter(pcollection)` instead.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_side_inputs_dict >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" custom_common_items >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 7: Combining with a `CombineFn`
+
+The more general way to combine elements, and the most flexible, is with a class that inherits from `CombineFn`.
+
+* [`CombineFn.create_accumulator()`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineFn.create_accumulator):
+ Called *once per `CombineFn` instance* when the `CombineFn` instance is initialized.
+ This creates an empty accumulator.
+ For example, an empty accumulator for a sum would be `0`, while an empty accumulator for a product (multiplication) would be `1`.
+
+* [`CombineFn.add_input()`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineFn.add_input):
+ Called *once per element*.
+ Takes an accumulator and an input element, combines them and returns the updated accumulator.
+
+* [`CombineFn.merge_accumulators()`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineFn.merge_accumulators):
+ Called *once per bundle of elements* after processing the last element of the bundle.
+ Multiple accumulators could be processed in parallel, so this function helps merging them into a single accumulator.
+
+* [`CombineFn.extract_output()`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineFn.extract_output):
Review comment:
"Called *once per `CombineFn` instance* when the `CombineFn` instance is done." -> this will be runner dependent. I would drop this.
"After all accumulators have been merged into a single final accumulator, `extract_output` allows to do additional calculations." -> I will simplify this to drop the part about merging. This could be called without any merging (e.g. the case where there are not other accumulators to merge.)
"This is useful for calculating averages, percentages, or anything that needs aggregate information from all the elements." -> This describers combiners in general. Combiners need the whole interface including extract_output to to accomplish this goal.
##########
File path: website/www/site/content/en/documentation/transforms/python/aggregation/combineglobally.md
##########
@@ -14,29 +14,197 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
+
# CombineGlobally
-<table align="left">
- <a target="_blank" class="button"
- href="https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineGlobally">
- <img src="https://beam.apache.org/images/logos/sdks/python.png" width="20px" height="20px"
- alt="Pydoc" />
- Pydoc
- </a>
-</table>
-<br><br>
+{{< localstorage language language-py >}}
+{{< button-pydoc path="apache_beam.transforms.core" class="CombineGlobally" >}}
Combines all elements in a collection.
See more information in the [Beam Programming Guide](/documentation/programming-guide/#combine).
## Examples
-See [BEAM-7390](https://issues.apache.org/jira/browse/BEAM-7390) for updates.
-## Related transforms
+In the following examples, we create a pipeline with a `PCollection` of produce.
+Then, we apply `CombineGlobally` in multiple ways to combine all the elements in the `PCollection`.
+
+`CombineGlobally` accepts a function that takes a list of elements as an input, and combines them to return a single element.
+
+### Example 1: Combining with a function
+
+We define a function `get_common_items` which takes a list of sets as an input, and calculates the intersection (common items) of those sets.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_function >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 2: Combining with a lambda function
+
+We can also use lambda functions to simplify **Example 1**.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_lambda >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 3: Combining with multiple arguments
+
+You can pass functions with multiple arguments to `CombineGlobally`.
+They are passed as additional positional arguments or keyword arguments to the function.
+
+In this example, the lambda function takes `sets` and `exclude` as arguments.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_multiple_arguments >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items_with_exceptions >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 4: Combining with side inputs as singletons
+
+If the `PCollection` has a single value, such as the average from another computation,
+passing the `PCollection` as a *singleton* accesses that value.
+
+In this example, we pass a `PCollection` the value `'🥕'` as a singleton.
+We then use that value to exclude specific items.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_side_inputs_singleton >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items_with_exceptions >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 5: Combining with side inputs as iterators
+
+If the `PCollection` has multiple values, pass the `PCollection` as an *iterator*.
+This accesses elements lazily as they are needed,
+so it is possible to iterate over large `PCollection`s that won't fit into memory.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_side_inputs_iter >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items_with_exceptions >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+> **Note**: You can pass the `PCollection` as a *list* with `beam.pvalue.AsList(pcollection)`,
+> but this requires that all the elements fit into memory.
+
+### Example 6: Combining with side inputs as dictionaries
+
+If a `PCollection` is small enough to fit into memory, then that `PCollection` can be passed as a *dictionary*.
+Each element must be a `(key, value)` pair.
+Note that all the elements of the `PCollection` must fit into memory for this.
+If the `PCollection` won't fit into memory, use `beam.pvalue.AsIter(pcollection)` instead.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_side_inputs_dict >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" custom_common_items >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 7: Combining with a `CombineFn`
+
+The more general way to combine elements, and the most flexible, is with a class that inherits from `CombineFn`.
+
+* [`CombineFn.create_accumulator()`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineFn.create_accumulator):
+ Called *once per `CombineFn` instance* when the `CombineFn` instance is initialized.
+ This creates an empty accumulator.
+ For example, an empty accumulator for a sum would be `0`, while an empty accumulator for a product (multiplication) would be `1`.
+
+* [`CombineFn.add_input()`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineFn.add_input):
+ Called *once per element*.
+ Takes an accumulator and an input element, combines them and returns the updated accumulator.
+
+* [`CombineFn.merge_accumulators()`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineFn.merge_accumulators):
+ Called *once per bundle of elements* after processing the last element of the bundle.
Review comment:
> @aaltay can you confirm if this and the rest of the descriptions are correct? Thanks!
Will do.
There is no guarantee that `merge_accumulators` will be called once per bundle. It could be called more than once depending on the runner implementation.
----------------------------------------------------------------
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] davidcavazos commented on a change in pull request #12252: [BEAM-7390] Add combineglobally code snippets
Posted by GitBox <gi...@apache.org>.
davidcavazos commented on a change in pull request #12252:
URL: https://github.com/apache/beam/pull/12252#discussion_r454491828
##########
File path: website/www/site/content/en/documentation/transforms/python/aggregation/combineglobally.md
##########
@@ -14,29 +14,197 @@ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
+
# CombineGlobally
-<table align="left">
- <a target="_blank" class="button"
- href="https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineGlobally">
- <img src="https://beam.apache.org/images/logos/sdks/python.png" width="20px" height="20px"
- alt="Pydoc" />
- Pydoc
- </a>
-</table>
-<br><br>
+{{< localstorage language language-py >}}
+{{< button-pydoc path="apache_beam.transforms.core" class="CombineGlobally" >}}
Combines all elements in a collection.
See more information in the [Beam Programming Guide](/documentation/programming-guide/#combine).
## Examples
-See [BEAM-7390](https://issues.apache.org/jira/browse/BEAM-7390) for updates.
-## Related transforms
+In the following examples, we create a pipeline with a `PCollection` of produce.
+Then, we apply `CombineGlobally` in multiple ways to combine all the elements in the `PCollection`.
+
+`CombineGlobally` accepts a function that takes a list of elements as an input, and combines them to return a single element.
+
+### Example 1: Combining with a function
+
+We define a function `get_common_items` which takes a list of sets as an input, and calculates the intersection (common items) of those sets.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_function >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 2: Combining with a lambda function
+
+We can also use lambda functions to simplify **Example 1**.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_lambda >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 3: Combining with multiple arguments
+
+You can pass functions with multiple arguments to `CombineGlobally`.
+They are passed as additional positional arguments or keyword arguments to the function.
+
+In this example, the lambda function takes `sets` and `exclude` as arguments.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_multiple_arguments >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items_with_exceptions >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 4: Combining with side inputs as singletons
+
+If the `PCollection` has a single value, such as the average from another computation,
+passing the `PCollection` as a *singleton* accesses that value.
+
+In this example, we pass a `PCollection` the value `'🥕'` as a singleton.
+We then use that value to exclude specific items.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_side_inputs_singleton >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items_with_exceptions >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 5: Combining with side inputs as iterators
+
+If the `PCollection` has multiple values, pass the `PCollection` as an *iterator*.
+This accesses elements lazily as they are needed,
+so it is possible to iterate over large `PCollection`s that won't fit into memory.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_side_inputs_iter >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" common_items_with_exceptions >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+> **Note**: You can pass the `PCollection` as a *list* with `beam.pvalue.AsList(pcollection)`,
+> but this requires that all the elements fit into memory.
+
+### Example 6: Combining with side inputs as dictionaries
+
+If a `PCollection` is small enough to fit into memory, then that `PCollection` can be passed as a *dictionary*.
+Each element must be a `(key, value)` pair.
+Note that all the elements of the `PCollection` must fit into memory for this.
+If the `PCollection` won't fit into memory, use `beam.pvalue.AsIter(pcollection)` instead.
+
+{{< highlight py >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" combineglobally_side_inputs_dict >}}
+{{< /highlight >}}
+
+{{< paragraph class="notebook-skip" >}}
+Output `PCollection` after `CombineGlobally`:
+{{< /paragraph >}}
+
+{{< highlight class="notebook-skip" >}}
+{{< code_sample "sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally_test.py" custom_common_items >}}
+{{< /highlight >}}
+
+{{< buttons-code-snippet
+ py="sdks/python/apache_beam/examples/snippets/transforms/aggregation/combineglobally.py" >}}
+
+### Example 7: Combining with a `CombineFn`
+
+The more general way to combine elements, and the most flexible, is with a class that inherits from `CombineFn`.
+
+* [`CombineFn.create_accumulator()`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineFn.create_accumulator):
+ Called *once per `CombineFn` instance* when the `CombineFn` instance is initialized.
+ This creates an empty accumulator.
+ For example, an empty accumulator for a sum would be `0`, while an empty accumulator for a product (multiplication) would be `1`.
+
+* [`CombineFn.add_input()`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineFn.add_input):
+ Called *once per element*.
+ Takes an accumulator and an input element, combines them and returns the updated accumulator.
+
+* [`CombineFn.merge_accumulators()`](https://beam.apache.org/releases/pydoc/current/apache_beam.transforms.core.html#apache_beam.transforms.core.CombineFn.merge_accumulators):
+ Called *once per bundle of elements* after processing the last element of the bundle.
Review comment:
@aaltay can you confirm if this and the rest of the descriptions are correct? Thanks!
----------------------------------------------------------------
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] davidcavazos edited a comment on pull request #12252: [BEAM-7390] Add combineglobally code snippets
Posted by GitBox <gi...@apache.org>.
davidcavazos edited a comment on pull request #12252:
URL: https://github.com/apache/beam/pull/12252#issuecomment-658939252
I also noticed that the output tags were a little misleading now since we are showing the stdout instead of an output PCollection.
Opened #12269 for that
----------------------------------------------------------------
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] davidcavazos commented on pull request #12252: [BEAM-7390] Add combineglobally code snippets
Posted by GitBox <gi...@apache.org>.
davidcavazos commented on pull request #12252:
URL: https://github.com/apache/beam/pull/12252#issuecomment-658891683
Staged: http://apache-beam-website-pull-requests.storage.googleapis.com/12252/documentation/transforms/python/aggregation/combineglobally/index.html
----------------------------------------------------------------
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] davidcavazos commented on pull request #12252: [BEAM-7390] Add combineglobally code snippets
Posted by GitBox <gi...@apache.org>.
davidcavazos commented on pull request #12252:
URL: https://github.com/apache/beam/pull/12252#issuecomment-658937924
@aaltay I've addressed the comments
----------------------------------------------------------------
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] aaltay commented on pull request #12252: [BEAM-7390] Add combineglobally code snippets
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #12252:
URL: https://github.com/apache/beam/pull/12252#issuecomment-658929492
(Please ping me once the open comments are addressed.)
----------------------------------------------------------------
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