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/06/25 02:05:41 UTC
[GitHub] [beam] aaltay opened a new pull request #12088: Add output typehints to GroupIntoBatches
aaltay opened a new pull request #12088:
URL: https://github.com/apache/beam/pull/12088
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark
--- | --- | --- | --- | --- | ---
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/)
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/) | [](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_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] udim commented on a change in pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r445278714
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
I was thinking it might be Iterable rather than List, so I looked at the code and it doesn't seem to group by key at all.
I modified a test from `GroupIntoBatchesTest` to print to elements:
```py
def test_in_global_window(self):
with TestPipeline() as pipeline:
def print_elements(e):
print(e)
return e
collection = pipeline \
| beam.Create(GroupIntoBatchesTest._create_test_data()) \
| util.GroupIntoBatches(GroupIntoBatchesTest.BATCH_SIZE) \
| beam.Map(print_elements)
```
Printout:
```py
[('key', 'Einstein'), ('key', 'Darwin'), ('key', 'Copernicus'), ('key', 'Pasteur'), ('key', 'Curie')]
[('key', 'Faraday'), ('key', 'Newton'), ('key', 'Bohr'), ('key', 'Galilei'), ('key', 'Maxwell')]
```
----------------------------------------------------------------
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] udim commented on a change in pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r445927300
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
The Java implementation splits (key, value) pairs.
https://github.com/apache/beam/blob/eaa41cc4cbcc4f94d0ec1a36ff2b0f3fcee962f9/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java#L173-L174
I don't see that in Python - is the runner supposed to do 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] udim commented on a change in pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r445278906
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
cc: @lukecwik
----------------------------------------------------------------
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 #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
aaltay commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r445756416
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
I wonder why the `test_in_global_window` tests passes. The assert following that code is seemingly checking for the right batching.
The implementation matches the Java one (https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java#L70). I _think_ grouping by key is supposed to happen as part of the state api, because element_state will be keyed by key and window.
----------------------------------------------------------------
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] kennknowles commented on a change in pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r445894269
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
A very real situation is that it might already be partitioned however the runner wants, via analyzing the pipeline.
----------------------------------------------------------------
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] kennknowles commented on a change in pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
kennknowles commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r445894084
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
Correct. It is up to the runner how to implement, but per key and window must be executed serially.
----------------------------------------------------------------
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 #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
aaltay commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r447358864
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
I think I understand the difference between Python and Java.
In Java, this works by emitting (Key, Iterable of Values) (here: https://github.com/apache/beam/blob/51bd3a4588984c9680f61a3268513fe5bde3ba96/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java#L212)
In Python, it stores (K, V) pairs elements (here: https://github.com/apache/beam/blob/51bd3a4588984c9680f61a3268513fe5bde3ba96/sdks/python/apache_beam/transforms/util.py#L784) and emits them as a list (here: https://github.com/apache/beam/blob/51bd3a4588984c9680f61a3268513fe5bde3ba96/sdks/python/apache_beam/transforms/util.py#L800) result in a List[Tuple[Key, Value]]
If my understanding is correct, I can change the Python behavior to be similar to the Java one. And I can also add the space sharing optimization.
----------------------------------------------------------------
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 closed pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
aaltay closed pull request #12088:
URL: https://github.com/apache/beam/pull/12088
----------------------------------------------------------------
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 #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
aaltay commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r445895986
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
Thank you. This might be a bug with whatever runner Udi used (one of the direct runners). I am still puzzled by what would be a good implementation of this case in batch. Would a naive implementation be always insert a shuffle, and use an memory backed state store for the batch 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] aaltay commented on pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
aaltay commented on pull request #12088:
URL: https://github.com/apache/beam/pull/12088#issuecomment-652559154
Closing this PR. https://github.com/apache/beam/pull/12129 supersedes this.
----------------------------------------------------------------
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] udim commented on a change in pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r447179875
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
So is grouping by key being done at all by this transform? If so, how?
And why isn't the test getting a grouped result?
----------------------------------------------------------------
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] udim commented on a change in pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r445925136
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
The test passes because 2 batches/elements are output. And I was running the test via pytest as usual.
I don't follow the state api reasoning, and this is not documented in GroupIntoBatches. If you get a different result in DirectRunner vs DataflowRunner, it's a bug.
----------------------------------------------------------------
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 #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
aaltay commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r447379446
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
I am thinking something like this: https://github.com/apache/beam/pull/12129 -- Have not really tested this yet beyond a unit test. I ran the the print test Udi wrote above and the output is batched by key.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r447128500
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
> The Java implementation splits (key, value) pairs.
> https://github.com/apache/beam/blob/eaa41cc4cbcc4f94d0ec1a36ff2b0f3fcee962f9/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupIntoBatches.java#L173-L174
>
> I don't see that in Python - is the runner supposed to do that?
This is a space saving optimization since the key is always going to be the same. Instead of having a bag of (<K, V1>, <K, V2>, <K, V3>, ...) we use two state cells, one storing K and the other storing (V1, V2, V3, ...) and combine them to recreate the output. It would be worthwhile to replicate this optimization in Python as well but it isn't necessary from a correctness point of view.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #12088: Add output typehints to GroupIntoBatches
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #12088:
URL: https://github.com/apache/beam/pull/12088#discussion_r447122816
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -741,6 +741,7 @@ def WithKeys(pcoll, k):
@experimental()
@typehints.with_input_types(Tuple[K, V])
+@typehints.with_output_types(Tuple[K, List[V]])
Review comment:
> Thank you. This might be a bug with whatever runner Udi used (one of the direct runners). I am still puzzled by what would be a good implementation of this case in batch. Would a naive implementation be always insert a shuffle, and use an memory backed state store for the batch case?
Using an in memory representation is exactly what is done for all Beam runners today and can only be done because these runners execute the whole stage for a single key within a single bundle.
----------------------------------------------------------------
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