robertwb commented on code in PR #41054:
URL: https://github.com/apache/beam/pull/41054#discussion_r4544896365
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
+ """
+ def expand(self, pcoll):
+ raise NotImplementedError
+
+ def _expand(self, pcoll, post_process):
Review Comment:
Done.
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
+ """
+ def expand(self, pcoll):
+ raise NotImplementedError
+
+ def _expand(self, pcoll, post_process):
+ if pcoll.windowing.is_default():
Review Comment:
This now fails as before.
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
+ """
+ def expand(self, pcoll):
+ raise NotImplementedError
+
+ def _expand(self, pcoll, post_process):
+ if pcoll.windowing.is_default():
+ if self.has_defaults:
Review Comment:
Yes. Done.
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
+ """
+ def expand(self, pcoll):
+ raise NotImplementedError
+
+ def _expand(self, pcoll, post_process):
+ if pcoll.windowing.is_default():
+ if self.has_defaults:
+ return (
+ pcoll.pipeline
+ | 'DoOnce' >> core.Create([None])
+ | core.Map(lambda _, s: post_process(s), pvalue.AsIter(pcoll)))
Review Comment:
That's really sad... Added the workaround.
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
+ """
+ def expand(self, pcoll):
+ raise NotImplementedError
+
+ def _expand(self, pcoll, post_process):
+ if pcoll.windowing.is_default():
+ if self.has_defaults:
+ return (
+ pcoll.pipeline
+ | 'DoOnce' >> core.Create([None])
+ | core.Map(lambda _, s: post_process(s), pvalue.AsIter(pcoll)))
+ else:
+
+ def _if_non_empty(unused_elem, s):
+ for _ in s:
Review Comment:
Fixed.
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
+ """
+ def expand(self, pcoll):
+ raise NotImplementedError
+
+ def _expand(self, pcoll, post_process):
+ if pcoll.windowing.is_default():
+ if self.has_defaults:
+ return (
+ pcoll.pipeline
+ | 'DoOnce' >> core.Create([None])
+ | core.Map(lambda _, s: post_process(s), pvalue.AsIter(pcoll)))
+ else:
+
+ def _if_non_empty(unused_elem, s):
Review Comment:
Yep. Done.
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
Review Comment:
Yeah. ToDict is non-deterministic (which value wins) with overlapping keys
so I think that's not worth worrying about which is why I included it here.
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
+ """
+ def expand(self, pcoll):
+ raise NotImplementedError
+
+ def _expand(self, pcoll, post_process):
+ if pcoll.windowing.is_default():
+ if self.has_defaults:
+ return (
+ pcoll.pipeline
+ | 'DoOnce' >> core.Create([None])
+ | core.Map(lambda _, s: post_process(s), pvalue.AsIter(pcoll)))
+ else:
+
+ def _if_non_empty(unused_elem, s):
Review Comment:
_expand doesn't have the concrete type here, but I fixed this by suppressing
the inference at this level which will allow the declarations on the transform
itself to apply (and added tests).
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
+ """
+ def expand(self, pcoll):
+ raise NotImplementedError
+
+ def _expand(self, pcoll, post_process):
+ if pcoll.windowing.is_default():
Review Comment:
This is an odd corner case, but it now works as before.
##########
sdks/python/apache_beam/transforms/combiners.py:
##########
@@ -787,16 +789,71 @@ def add_input(self, accumulator, element, *args,
**kwargs):
]
+class ToSingleElementCollection(CombinerWithoutDefaults):
+ """Base class for combining a PCollection into a single element.
+
+ This is for combining a PCollection into a single element in the cases where
+ there is no value (and significant overhead) in using a combiner as the
+ output is no smaller than its input.
+ """
+ def expand(self, pcoll):
+ raise NotImplementedError
+
+ def _expand(self, pcoll, post_process):
+ if pcoll.windowing.is_default():
+ if self.has_defaults:
+ return (
+ pcoll.pipeline
+ | 'DoOnce' >> core.Create([None])
+ | core.Map(lambda _, s: post_process(s), pvalue.AsIter(pcoll)))
+ else:
+
+ def _if_non_empty(unused_elem, s):
Review Comment:
Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
