claudevdm commented on code in PR #41054:
URL: https://github.com/apache/beam/pull/41054#discussion_r4528897688
##########
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:
The result of `without_defaults()` now carries the minimum timestamp instead
of the end of the global window.
```python
import apache_beam as beam
with beam.Pipeline() as p:
_ = (
p
| beam.Create([1, 2, 3])
| beam.combiners.ToList().without_defaults()
| beam.Map(lambda x, ts=beam.DoFn.TimestampParam: print(sorted(x),
ts)))
# master: [1, 2, 3] Timestamp(9223371950454.775000)
# this PR: [1, 2, 3] Timestamp(-9223372036854.775000)
```
##########
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:
On Dataflow, a PCollection read with `AsIter` has its element type reset to
`Any` (`side_input_visitor`, #20043). That PCollection is now the user's own
input, so other steps reading it lose their typed coder as well.
```python
import apache_beam as beam
from apache_beam.runners.dataflow.dataflow_runner import DataflowRunner
p = beam.Pipeline()
kvs = p | beam.Create([('a', 1)])
_ = kvs | beam.combiners.ToDict()
_ = kvs | beam.GroupByKey()
print('before:', kvs.element_type)
# One of the rewrites DataflowRunner.run_pipeline applies before submitting.
p.visit(DataflowRunner.side_input_visitor())
print('after: ', kvs.element_type)
# master: after: Tuple[<class 'str'>, <class 'int'>]
# this PR: after: Any
```
With a key class that has its own coder, the `GroupByKey` then fails for me.
I applied the rewrites by hand and ran locally, I have not tried it on Dataflow.
<details>
<summary>Example that fails</summary>
```python
import apache_beam as beam
from apache_beam.runners.dataflow.dataflow_runner import DataflowRunner
from apache_beam.runners.pipeline_utils import group_by_key_input_visitor
class MyKey:
def __init__(self, v):
self.v = v
def __eq__(self, other):
return self.v == other.v
def __hash__(self):
return hash(self.v)
class MyKeyCoder(beam.coders.Coder):
def encode(self, key):
return str(key.v).encode()
def decode(self, data):
return MyKey(int(data.decode()))
def is_deterministic(self):
return True
beam.coders.registry.register_coder(MyKey, MyKeyCoder)
p = beam.Pipeline()
kvs = p | beam.Create([(MyKey(1), 1),
(MyKey(2), 2)]).with_output_types(tuple[MyKey, int])
_ = kvs | beam.combiners.ToDict()
_ = kvs | beam.GroupByKey()
# The rewrites DataflowRunner.run_pipeline applies before submitting.
p.visit(DataflowRunner.side_input_visitor())
p.visit(group_by_key_input_visitor())
p.run().wait_until_finish()
# master: runs fine
# this PR: Unable to deterministically encode '<MyKey object>' ... for the
input of 'GroupByKey'
```
</details>
Would it make sense to side-input a copy (`pcoll | Map(lambda x: x)`) so the
user's PCollection is left alone?
##########
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:
This is true for `ToList`. For `ToDict` with repeated keys, the old combiner
folded each bundle into one small dict before the shuffle. Now every pair goes
through.
```python
import time
import apache_beam as beam
def cpu_seconds(num_keys):
start = time.process_time()
with beam.Pipeline('FnApiRunner') as p:
_ = (
p
| beam.Create(range(300_000))
| beam.Map(lambda x: (x % num_keys, x))
| beam.combiners.ToDict())
return time.process_time() - start
print('1,000 distinct keys: %.1fs CPU' % cpu_seconds(1000))
print('300,000 distinct keys: %.1fs CPU' % cpu_seconds(300_000))
# master: 1,000 keys ~15s 300,000 keys ~19s
# this PR: 1,000 keys ~18.5s 300,000 keys ~19s
```
This is the in-process runner, so it won't show the shuffle savings you are
after. It only shows the repeated-key case doing more work than before. Do you
think that case is rare enough to ignore, or should `ToDict` stay on the
combiner?
##########
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:
A global window with a custom trigger used to work with `ToList()` in batch.
Now it raises.
```python
import apache_beam as beam
from apache_beam.transforms import trigger
from apache_beam.transforms import window
with beam.Pipeline() as p:
_ = (
p
| beam.Create([1, 2, 3])
| beam.WindowInto(
window.GlobalWindows(),
trigger=trigger.Repeatedly(trigger.AfterCount(2)),
accumulation_mode=trigger.AccumulationMode.DISCARDING)
| beam.combiners.ToList()
| beam.Map(print))
# master: [1, 2, 3]
# this PR: ValueError: Default values are not yet supported in ToList() if
the
# output PCollection is not windowed by GlobalWindows. ...
```
Is that intended?
##########
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:
1. This works on master in batch and raises here:
```python
pcoll | WindowInto(GlobalWindows(), trigger=Repeatedly(AfterCount(2)),
accumulation_mode=AccumulationMode.DISCARDING) | ToList()
```
Is that intended? The message also says the input is "not windowed by
GlobalWindows", which it is.
2. `PeriodicImpulse(fire_interval=1) | ToList()` fails at construction on
master (the GroupByKey check on unbounded input). Here it builds and, as you'd
expect, emits nothing until the job is drained. Do we want to keep failing
early for this, or is waiting quietly fine?
##########
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:
With `without_defaults()` the output type is now `List[Any]` instead of
`List[int]` (and `Union[]` for `ToIterable`), so the typed coder and downstream
type checks are lost. I think it's because the data comes in through the side
input, which type inference can't see. Could `_expand` declare the real output
type on the inner steps, meaning the transform's own `list[T]` bound to the
input's element type?
##########
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:
On an unbounded input, master stops you at construction. This PR builds the
pipeline, and I think it would then output nothing until drain, because the
side input is only ready when the global window closes.
```python
import apache_beam as beam
from apache_beam.transforms.periodicsequence import PeriodicImpulse
p = beam.Pipeline()
_ = p | PeriodicImpulse(fire_interval=1) | beam.combiners.ToList()
# master: ValueError: GroupByKey cannot be applied to an unbounded
PCollection
# with global windowing and a default trigger
# this PR: no error
```
Should we keep an early error for this case?
##########
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:
With `without_defaults()` the output loses the element type, so the elements
fall back to the generic coder.
```python
import apache_beam as beam
p = beam.Pipeline()
ints = p | beam.Create([1, 2, 3])
print((ints | 'A' >> beam.combiners.ToList()).element_type)
print((ints | 'B' >>
beam.combiners.ToList().without_defaults()).element_type)
print((ints | 'C' >>
beam.combiners.ToIterable().without_defaults()).element_type)
# master (first two lines):
# List[<class 'int'>]
# List[<class 'int'>]
# this PR:
# List[<class 'int'>]
# List[Any]
# Union[]
```
Would `.with_output_types(Any)` on the FlatMap and MapTuple fix it? That
seems to be why the defaults path still gets `List[int]`.
##########
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:
This breaks streaming compatibility "The new job is missing steps
ToList/ToList/CombinePerKey/GroupByKey". Mapping that step did not help: "The
stage ToList/GroupByKey has changed in a way that is incompatible with update".
Could we keep the old expansion behind `is_compat_version_prior_to(...)`,
like Reshuffle does, and add a CHANGES.md note? Also, `Map(<lambda at
combiners.py:808>)` reads the side input, and renaming a side-input reader
breaks updates too, so could it get a fixed label?
##########
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:
A streaming job is updated by matching step names, and the names inside
`ToList`/`ToDict` all change here. One of the new names also contains a line
number, so it would change again whenever this file is edited.
```python
import apache_beam as beam
from apache_beam.transforms import window
p = beam.Pipeline()
_ = (
p
| beam.Create([1, 2, 3])
| beam.WindowInto(window.FixedWindows(60))
| beam.combiners.ToList().without_defaults())
for t in p.to_runner_api().components.transforms.values():
if t.unique_name.startswith('ToList/') and not t.subtransforms:
print(t.unique_name)
# master:
# ToList/ToList/KeyWithVoid
# ToList/ToList/CombinePerKey/GroupByKey
# ToList/ToList/CombinePerKey/Combine/ParDo(CombineValuesDoFn)
# ToList/ToList/UnKey
# this PR:
# ToList/KeyWithVoid
# ToList/GroupByKey
# ToList/MapTuple(<lambda at combiners.py:832>)
```
Could we keep the old expansion behind `is_compat_version_prior_to(...)`,
like Reshuffle does? And give the lambdas explicit labels, like `Create` does
for its own (#36381)? A CHANGES.md note might help too.
##########
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:
On Dataflow, a PCollection read with `AsIter` gets its element type reset to
`Any` (`side_input_visitor`, #20043). That is now the user's own input, so
other steps reading it lose their coder. This passes on master and fails here
on Dataflow:
```python
rows = p | beam.Create([beam.Row(a=1, b='x'), beam.Row(a=2, b='y')])
_ = rows | beam.combiners.ToList()
_ = rows | SqlTransform('SELECT a, b FROM PCOLLECTION WHERE a > 1')
# CoderException: `UnknownCoderWrapper` was used to perform an actual
decoding in the Java SDK
```
A `GroupByKey` keyed by a class with its own coder fails the same way. The
reset looks like a leftover from when Dataflow got both a JSON and a proto job
description?When I removed it locally, both pipelines passed on Dataflow.
##########
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:
True for `ToList`, but `ToDict` with repeated keys used to shrink in the
combiner.
##########
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:
`ToList().without_defaults()` now outputs at the minimum timestamp instead
of the end of the global window. That matches what the defaults path already
did, so it seems fine, but it could move the result if someone windows it
afterwards. Worth a CHANGES.md note?
--
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]
