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/12/05 23:45:24 UTC
[GitHub] [beam] nehsyc opened a new pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
nehsyc opened a new pull request #13493:
URL: https://github.com/apache/beam/pull/13493
Specify output type for `GroupIntoBatches.WithShardedKey` using `ShardedKeyTypeConstraint` so during the runner API translation the corresponding coder can be chosen.
------------------------
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_VR_Dataflow_V2/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_Dataflow/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] nehsyc commented on pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13493:
URL: https://github.com/apache/beam/pull/13493#issuecomment-739589799
CC: @boyuanzz @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] udim commented on a change in pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538775475
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
Right, Python's `typing` types don't recognize Beam's own internal types.
1.
Probably not a solution:
ShardedKeyTypeConstraint could inherit from `typing.Generic[typing.TypeVar('K')]`, but that produces 2 more problems. Neither are hard but add complexity, plus none of the other Beam typehints do this.
- Accessing the type argument that Generic stores on different versions of Python. (`__args__` or `__parameters__` I believe)
- Pickling. Python's typing types have had issues with pickling.
2.
Probably the right solution but breaks consistency with the other decorators:
```py
@typehints.with_output_types(typehints.Tuple[ShardedKeyType[K], typehints.Iterable[V]])
```
Beam already converts to internal type so this is a cosmetic change.
----------------------------------------------------------------
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] boyuanzz commented on a change in pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r537903184
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
Just curious, does `@typehints.with_output_types` not work here?
----------------------------------------------------------------
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] nehsyc commented on pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13493:
URL: https://github.com/apache/beam/pull/13493#issuecomment-742856731
cc @pabloem
----------------------------------------------------------------
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 #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538775475
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
Right, Python's `typing` types don't recognize Beam's own internal types.
1. Probably not a solution:
ShardedKeyTypeConstraint could inherit from `typing.Generic[typing.TypeVar('K')]`, but that produces 2 more problems. Neither are hard but add complexity, plus none of the other Beam typehints do this.
- Accessing the type argument that Generic stores on different versions of Python. (`__args__` or `__parameters__` I believe)
- Pickling. Python's typing types have had issues with pickling.
2. Probably the right solution but breaks consistency with the other decorators:
```py
@typehints.with_output_types(typehints.Tuple[ShardedKeyType[K], typehints.Iterable[V]])
```
Beam already converts to internal type so this is a cosmetic change.
----------------------------------------------------------------
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] nehsyc commented on pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13493:
URL: https://github.com/apache/beam/pull/13493#issuecomment-741558098
This PR is likely going to be conflicted with #13405. Let me know if it looks good and I can pull the head and merge.
----------------------------------------------------------------
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 #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538958762
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
I don't believe you need infer_output_type if you have the with_output_types decorator.
Also, I tested this a bit myself and it ended up looking like this:
```py
@typehints.with_input_types(typehints.Tuple[K_i, V_i])
@typehints.with_output_types(typehints.Tuple[ShardedKeyType[K_i], typehints.Iterable[V_i]])
class WithShardedKey(PTransform):
```
where:
```py
K_i = typehints.TypeVariable('K_i')
V_i = typehints.TypeVariable('V_i')
```
edit: not sure if `with_input_types` needs to change but it looks consistent
----------------------------------------------------------------
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 #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538958762
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
I don't believe you need infer_output_type if you have the with_output_types decorator.
Also, I tested this a bit myself and it ended up looking like this:
```py
@typehints.with_input_types(typehints.Tuple[K_i, V_i])
@typehints.with_output_types(typehints.Tuple[ShardedKeyType[K_i], typehints.Iterable[V_i]])
class WithShardedKey(PTransform):
```
where:
```py
K_i = typehints.TypeVariable('K_i')
V_i = typehints.TypeVariable('V_i')
```
----------------------------------------------------------------
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 #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538729953
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -815,18 +817,25 @@ def __init__(self, batch_size, max_buffering_duration_secs=None):
_shard_id_prefix = uuid.uuid4().bytes
def expand(self, pcoll):
+ key_type, value_type = pcoll.element_type.tuple_types
sharded_pcoll = pcoll | Map(
lambda key_value: (
ShardedKey(
key_value[0],
# Use [uuid, thread id] as the shard id.
GroupIntoBatches.WithShardedKey._shard_id_prefix + bytes(
threading.get_ident().to_bytes(8, 'big'))),
- key_value[1]))
+ key_value[1])).with_output_types(
+ TupleConstraint([ShardedKeyType[key_type], value_type]))
Review comment:
Sorry, my mistake I didn't read this carefully enough.
----------------------------------------------------------------
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 #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538762100
##########
File path: sdks/python/apache_beam/typehints/sharded_key_type.py
##########
@@ -25,8 +25,12 @@
from apache_beam.typehints.typehints import match_type_variables
from apache_beam.utils.sharded_key import ShardedKey
+from future.utils import with_metaclass
-class ShardedKeyTypeConstraint(typehints.TypeConstraint):
+
+class ShardedKeyTypeConstraint(with_metaclass(typehints.GetitemConstructor,
Review comment:
This works:
```py
def __class_getitem__(cls, py_type):
return cls(py_type)
```
The alternative is to use a wrapper object instance such as `IterableHint` that has `__getitem__`, and then define `Iterable = IterableHint()`.
----------------------------------------------------------------
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] robinyqiu merged pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
robinyqiu merged pull request #13493:
URL: https://github.com/apache/beam/pull/13493
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538915141
##########
File path: sdks/python/apache_beam/typehints/sharded_key_type.py
##########
@@ -25,8 +25,12 @@
from apache_beam.typehints.typehints import match_type_variables
from apache_beam.utils.sharded_key import ShardedKey
+from future.utils import with_metaclass
-class ShardedKeyTypeConstraint(typehints.TypeConstraint):
+
+class ShardedKeyTypeConstraint(with_metaclass(typehints.GetitemConstructor,
Review comment:
Thanks for the tip!!! Updated the code to define `__class_getitem__` directly in `ShardedKeyTypeConstraint`.
----------------------------------------------------------------
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 #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538959518
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -72,6 +72,9 @@
from apache_beam.transforms.window import NonMergingWindowFn
from apache_beam.transforms.window import TimestampCombiner
from apache_beam.transforms.window import TimestampedValue
+from apache_beam.typehints.sharded_key_type import ShardedKeyType
+from apache_beam.typehints.typehints import IterableTypeConstraint
+from apache_beam.typehints.typehints import TupleConstraint
Review comment:
Okay, this is fine then :)
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r539038111
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
Sorry I misunderstood and thought having something like `typehints.Tuple` in the decoration was not preferred. Pushed a commit to use decoration instead.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538631115
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -815,18 +817,25 @@ def __init__(self, batch_size, max_buffering_duration_secs=None):
_shard_id_prefix = uuid.uuid4().bytes
def expand(self, pcoll):
+ key_type, value_type = pcoll.element_type.tuple_types
sharded_pcoll = pcoll | Map(
lambda key_value: (
ShardedKey(
key_value[0],
# Use [uuid, thread id] as the shard id.
GroupIntoBatches.WithShardedKey._shard_id_prefix + bytes(
threading.get_ident().to_bytes(8, 'big'))),
- key_value[1]))
+ key_value[1])).with_output_types(
+ TupleConstraint([ShardedKeyType[key_type], value_type]))
return (
sharded_pcoll
| GroupIntoBatches(self.batch_size, self.max_buffering_duration_secs))
+ def infer_output_type(self, input_type):
Review comment:
Couldn't use the decorator due to the error I mentioned in another comment.
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -815,18 +817,25 @@ def __init__(self, batch_size, max_buffering_duration_secs=None):
_shard_id_prefix = uuid.uuid4().bytes
def expand(self, pcoll):
+ key_type, value_type = pcoll.element_type.tuple_types
sharded_pcoll = pcoll | Map(
lambda key_value: (
ShardedKey(
key_value[0],
# Use [uuid, thread id] as the shard id.
GroupIntoBatches.WithShardedKey._shard_id_prefix + bytes(
threading.get_ident().to_bytes(8, 'big'))),
- key_value[1]))
+ key_value[1])).with_output_types(
+ TupleConstraint([ShardedKeyType[key_type], value_type]))
Review comment:
Why? The MapFn converts the input K, V to ShardedKey, V. The output of the composite transform is ShardedKey, Iterable[V].
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
Because I got an error saying:
```
arg = ShardedKey[int], msg = 'Tuple[t0, t1, ...]: each t must be a type.', is_argument = True
def _type_check(arg, msg, is_argument=True):
"""Check that the argument is a type, and return it (internal helper).
As a special case, accept None and return type(None) instead. Also wrap strings
into ForwardRef instances. Consider several corner cases, for example plain
special forms like Union are not valid, while Union[int, str] is OK, etc.
The msg argument is a human-readable error message, e.g::
"Union[arg, ...]: arg should be a type."
We append the repr() of the actual value (truncated to 100 chars).
"""
invalid_generic_forms = (Generic, Protocol)
if is_argument:
invalid_generic_forms = invalid_generic_forms + (ClassVar, Final)
if arg is None:
return type(None)
if isinstance(arg, str):
return ForwardRef(arg)
if (isinstance(arg, _GenericAlias) and
arg.__origin__ in invalid_generic_forms):
raise TypeError(f"{arg} is not valid as type argument")
if (isinstance(arg, _SpecialForm) and arg not in (Any, NoReturn) or
arg in (Generic, Protocol)):
raise TypeError(f"Plain {arg} is not valid as type argument")
if isinstance(arg, (type, TypeVar, ForwardRef)):
return arg
if not callable(arg):
> raise TypeError(f"{msg} Got {arg!r:.100}.")
E TypeError: Tuple[t0, t1, ...]: each t must be a type. Got ShardedKey[int].
../../../../../../.pyenv/versions/3.8.5/lib/python3.8/typing.py:149: TypeError
```
Any insights?
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
Udi also had a similar question. I got TypeError when trying to use Tuple[ShardedKey,...]. See my reply below.
##########
File path: sdks/python/apache_beam/typehints/sharded_key_type.py
##########
@@ -25,8 +25,12 @@
from apache_beam.typehints.typehints import match_type_variables
from apache_beam.utils.sharded_key import ShardedKey
+from future.utils import with_metaclass
-class ShardedKeyTypeConstraint(typehints.TypeConstraint):
+
+class ShardedKeyTypeConstraint(with_metaclass(typehints.GetitemConstructor,
Review comment:
`GetItemConstructor` basically defines `__getitem__`. I tried getting rid of it and defining `__getitem__` directly but I got the error
```
TypeError: 'type' object is not subscriptable
```
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538914797
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -72,6 +72,9 @@
from apache_beam.transforms.window import NonMergingWindowFn
from apache_beam.transforms.window import TimestampCombiner
from apache_beam.transforms.window import TimestampedValue
+from apache_beam.typehints.sharded_key_type import ShardedKeyType
+from apache_beam.typehints.typehints import IterableTypeConstraint
+from apache_beam.typehints.typehints import TupleConstraint
Review comment:
Updated the code to use `Tuple` and `Iterable` directly. Couldn't add `ShardedKeyType` to `apache_beam/typehints/__init__.py` due to circular imports in coders
```
ImportError while loading conftest '/usr/local/google/home/sychen/Documents/GitHub/working_dir/beam/sdks/python/conftest.py'.
conftest.py:23: in <module>
from apache_beam.options import pipeline_options
apache_beam/__init__.py:95: in <module>
from apache_beam import coders
apache_beam/coders/__init__.py:19: in <module>
from apache_beam.coders.coders import *
apache_beam/coders/coders.py:52: in <module>
from apache_beam.typehints import typehints
apache_beam/typehints/__init__.py:25: in <module>
from apache_beam.typehints.sharded_key_type import ShardedKeyType
apache_beam/typehints/sharded_key_type.py:22: in <module>
from apache_beam.coders import typecoders
apache_beam/coders/typecoders.py:81: in <module>
from apache_beam.coders.coders import CoderElementType
E ImportError: cannot import name 'CoderElementType' from partially initialized module 'apache_beam.coders.coders' (most likely due to a circular import) (/usr/local/google/home/sychen/Documents/GitHub/working_dir/beam/sdks/python/apache_beam/coders/coders.py)
```
----------------------------------------------------------------
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 pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on pull request #13493:
URL: https://github.com/apache/beam/pull/13493#issuecomment-742150910
> hmm seems like python 3.6 failed and complained about
>
> ```
> File "/home/runner/work/beam/beam/sdks/python/target/.tox/py36/lib/python3.6/site-packages/apache_beam/transforms/util.py", line 803, in GroupIntoBatches
> ShardedKeyType[typehints.TypeVariable(K)], # type: ignore[misc]
> TypeError: 'type' object is not subscriptable
> ```
It looks like Python 3.6 doesn't support `__class_getitem__`, only 3.7+. Usually the docs state this but not in this case...
I think using the metaclass is the only solution that works on all versions.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r537156342
##########
File path: sdks/python/apache_beam/typehints/sharded_key_type.py
##########
@@ -25,8 +25,12 @@
from apache_beam.typehints.typehints import match_type_variables
from apache_beam.utils.sharded_key import ShardedKey
+from future.utils import with_metaclass
-class ShardedKeyTypeConstraint(typehints.TypeConstraint):
+
+class ShardedKeyTypeConstraint(with_metaclass(typehints.GetitemConstructor,
Review comment:
This seems to be casuing lint failure. I was following the `WindowedValue` example:
https://github.com/apache/beam/blob/30f9a607509940f78459e4fba847617399780246/sdks/python/apache_beam/typehints/typehints.py#L1050
----------------------------------------------------------------
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] nehsyc commented on pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13493:
URL: https://github.com/apache/beam/pull/13493#issuecomment-742102323
@udim
hmm seems like python 3.6 failed and complained about
```
File "/home/runner/work/beam/beam/sdks/python/target/.tox/py36/lib/python3.6/site-packages/apache_beam/transforms/util.py", line 803, in GroupIntoBatches
ShardedKeyType[typehints.TypeVariable(K)], # type: ignore[misc]
TypeError: 'type' object is not subscriptable
```
----------------------------------------------------------------
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] nehsyc commented on pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13493:
URL: https://github.com/apache/beam/pull/13493#issuecomment-742795645
> > hmm seems like python 3.6 failed and complained about
> > ```
> > File "/home/runner/work/beam/beam/sdks/python/target/.tox/py36/lib/python3.6/site-packages/apache_beam/transforms/util.py", line 803, in GroupIntoBatches
> > ShardedKeyType[typehints.TypeVariable(K)], # type: ignore[misc]
> > TypeError: 'type' object is not subscriptable
> > ```
>
> It looks like Python 3.6 doesn't support `__class_getitem__`, only 3.7+. Usually the docs state this but not in this case...
>
> I think using the metaclass is the only solution that works on all versions.
Python is hard :\ Kept getting `inherit-non-class` error with `with_metaclass` imported from `future.utils` and ended up using `six.with_metaclass` instead. Also added some doc string for `ShardedKeyTypeConstraint`.
All checks are passing now (finally!). It's ready to be merged.
----------------------------------------------------------------
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] nehsyc commented on pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13493:
URL: https://github.com/apache/beam/pull/13493#issuecomment-739589853
I didn't add `ShardedKeyTypeConstraint` to typehint.py specifically here
https://github.com/apache/beam/blob/30f9a607509940f78459e4fba847617399780246/sdks/python/apache_beam/typehints/typehints.py#L1116
due to cyclic imports. I could have moved the entire definition to typehints.py but I also needed to register the coder for the type constraint via `typecoders.registry.register_coder` which couldn't be done within typehints.py (again due to cyclic imports). Let me know if the current version is fine or any suggestions to make it more clear.
Thinking it more about whether this constraint can be used by users. Actually no such need. `ShardedKey` is not a magic type but just a normal type. We made it well-known because we want the transform `GroupIntoBatches` that uses `ShardedKey` to be a magic transform. So user defined transforms that involve `ShardedKey` don't have to have `ShardedKeyCoder` in the graph (generic fast primitive coder should work).
----------------------------------------------------------------
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 pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on pull request #13493:
URL: https://github.com/apache/beam/pull/13493#issuecomment-745500507
> Python is hard :\ Kept getting `inherit-non-class` error with `with_metaclass` imported from `future.utils` and ended up using `six.with_metaclass` instead. Also added some doc string for `ShardedKeyTypeConstraint`.
I would rather not use `six`, since it's for compatibility between Python 2.x and 3.x, and Beam doesn't support 2.x any more. I could help debug this further if you wish.
----------------------------------------------------------------
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] nehsyc commented on a change in pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r538914752
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
Ah that explains. To avoid the decorator inconsistency I still keep the `infer_output_type` method but instead use `typehint.Tuple` as opposed to `TupleConstraint` directly. Is that acceptable?
----------------------------------------------------------------
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 #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r539587103
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
It's not preferred, but I'd rather have the output type documented there than not at all.
----------------------------------------------------------------
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 #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
udim commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r537894802
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -72,6 +72,9 @@
from apache_beam.transforms.window import NonMergingWindowFn
from apache_beam.transforms.window import TimestampCombiner
from apache_beam.transforms.window import TimestampedValue
+from apache_beam.typehints.sharded_key_type import ShardedKeyType
+from apache_beam.typehints.typehints import IterableTypeConstraint
+from apache_beam.typehints.typehints import TupleConstraint
Review comment:
Please use:
```
from apache_beam.typehints import ShardedKeyType
```
And use `Tuple` and `Iterable` types from the typing module (they will get converted to the ones in typehints). In general use either `typing.Tuple` or `apache_beam.typehints.Tuple`. The constraint classes are internal to the typehints modules.
For the `ShardedKeyType` import to work, you can add the corresponding import statement to `apache_beam/typehints/__init__.py`. This also avoid the cyclic import you mentioned that happened with my suggestion to put ShardedKeyType in typehints.py.
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -785,7 +788,6 @@ def expand(self, pcoll):
@experimental()
@typehints.with_input_types(Tuple[K, V])
- @typehints.with_output_types(Tuple[K, Iterable[V]])
Review comment:
Any reason why this isn't:
```python
@typehints.with_output_types(Tuple[ShardedKeyType[K], Iterable[V]])
```
?
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -815,18 +817,25 @@ def __init__(self, batch_size, max_buffering_duration_secs=None):
_shard_id_prefix = uuid.uuid4().bytes
def expand(self, pcoll):
+ key_type, value_type = pcoll.element_type.tuple_types
sharded_pcoll = pcoll | Map(
lambda key_value: (
ShardedKey(
key_value[0],
# Use [uuid, thread id] as the shard id.
GroupIntoBatches.WithShardedKey._shard_id_prefix + bytes(
threading.get_ident().to_bytes(8, 'big'))),
- key_value[1]))
+ key_value[1])).with_output_types(
+ TupleConstraint([ShardedKeyType[key_type], value_type]))
Review comment:
If you keep this line, I believe the value_type should be `Iterable[value_type]`.
##########
File path: sdks/python/apache_beam/typehints/sharded_key_type.py
##########
@@ -25,8 +25,12 @@
from apache_beam.typehints.typehints import match_type_variables
from apache_beam.utils.sharded_key import ShardedKey
+from future.utils import with_metaclass
-class ShardedKeyTypeConstraint(typehints.TypeConstraint):
+
+class ShardedKeyTypeConstraint(with_metaclass(typehints.GetitemConstructor,
Review comment:
I don't know what `with_metaclass` does. You could add the `__getitem__` implementation directly if it's causing issues.
##########
File path: sdks/python/apache_beam/transforms/util.py
##########
@@ -815,18 +817,25 @@ def __init__(self, batch_size, max_buffering_duration_secs=None):
_shard_id_prefix = uuid.uuid4().bytes
def expand(self, pcoll):
+ key_type, value_type = pcoll.element_type.tuple_types
sharded_pcoll = pcoll | Map(
lambda key_value: (
ShardedKey(
key_value[0],
# Use [uuid, thread id] as the shard id.
GroupIntoBatches.WithShardedKey._shard_id_prefix + bytes(
threading.get_ident().to_bytes(8, 'big'))),
- key_value[1]))
+ key_value[1])).with_output_types(
+ TupleConstraint([ShardedKeyType[key_type], value_type]))
return (
sharded_pcoll
| GroupIntoBatches(self.batch_size, self.max_buffering_duration_secs))
+ def infer_output_type(self, input_type):
Review comment:
If you keep the with_output_types decorator, please remove 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] boyuanzz commented on a change in pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
boyuanzz commented on a change in pull request #13493:
URL: https://github.com/apache/beam/pull/13493#discussion_r537903634
##########
File path: sdks/python/apache_beam/typehints/sharded_key_type.py
##########
@@ -25,8 +25,12 @@
from apache_beam.typehints.typehints import match_type_variables
from apache_beam.utils.sharded_key import ShardedKey
+from future.utils import with_metaclass
-class ShardedKeyTypeConstraint(typehints.TypeConstraint):
+
+class ShardedKeyTypeConstraint(with_metaclass(typehints.GetitemConstructor,
Review comment:
What kind of lint errors are you getting?
----------------------------------------------------------------
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] nehsyc commented on pull request #13493: [BEAM-10475] Add output type hints for GroupIntoBatches.
Posted by GitBox <gi...@apache.org>.
nehsyc commented on pull request #13493:
URL: https://github.com/apache/beam/pull/13493#issuecomment-739589747
R: @udim
----------------------------------------------------------------
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