On Fri, Sep 15, 2023 at 9:46 AM Reuven Lax via user
<user@beam.apache.org> wrote:
Creating composite DoFns is tricky today due to how they are
implemented (via annotated methods).
Note that this depends on the language. This should be really easy
to do from Python.
However providing such a method to compose DoFns would be very
useful IMO.
+1
On Fri, Sep 15, 2023 at 9:33 AM Joey Tran
<joey.t...@schrodinger.com> wrote:
Yeah for (1) the concern would be adding a shuffle/fusion
break and (2) sounds like the likely solution, was just
hoping there'd be one that could wrap at the PTransform
level but I realize now the PTransform abstraction is too
general as you mentioned to do something like that.
(2) will be likely what we do, though now I'm wondering if
it might be possible to create a ParDo wrapper that can take
a ParDo, extract it's dofn, wrap it, and return a new ParDo
On Fri, Sep 15, 2023, 11:53 AM Robert Bradshaw via user
<user@beam.apache.org> wrote:
+1 to looking at composite transforms. You could even
have a composite transform that takes another transform
as one of its construction arguments and whose expand
method does pre- and post-processing to the
inputs/outputs before/after applying the transform in
question. (You could even implement this as a Python
decorator if you really wanted, either decorating the
expand method itself or the full class...)
One of the difficulties is that for a general transform
there isn't necessarily a 1:N relationship between
outputs and inputs as one has for a DoFn (especially if
there is any aggregation involved). There are, however,
two partial solutions that might help.
(1) You can do a CombineGlobally with a CombineFn (Like
Sample) that returns at most N elements. You could do
this with a CombinePerKey if you can come up with a
reasonable key (e.g. the id of your input elements) that
the limit should be a applied to. Note that this may
cause a lot of data to be shuffled (though due to
combiner lifting, no more than N per bundle).
(2) You could have a DoFn that limits to N per bundle by
initializing a counter in its start_bundle and passing
elements through until the counter reaches a threshold.
(Again, one could do this per id if one is available.)
It wouldn't stop production of the elements, but if
things get fused it would still likely be fairly cheap.
Both of these could be prepended to the problematic
consuming PTransform as well.
- Robert
On Fri, Sep 15, 2023 at 8:13 AM Joey Tran
<joey.t...@schrodinger.com> wrote:
I'm aware of composite transforms and of the
distributed nature of PTransforms. I'm not
suggesting limiting the entire set and my example
was more illustrative than the actual use case.
My actual use case is basically: I have multiple
PTransforms, and let's say most of them average ~100
generated outputs for a single input. Most of these
PTransforms will occasionally run into an input
though that might output maybe 1M outputs. This can
cause issues if for example there are transforms
that follow it that require a lot of compute per input.
The simplest way to deal with this is to modify the
`DoFn`s in our Ptransforms and add a limiter in the
logic (e.g. `if num_outputs_generated >=
OUTPUTS_PER_INPUT_LIMIT: return`). We could
duplicate this logic across our transforms, but it'd
be much cleaner if we could lift up this limiting
logic out of the application logic and have some
generic wrapper that extends our transforms.
Thanks for the discussion!
On Fri, Sep 15, 2023 at 10:29 AM Alexey Romanenko
<aromanenko....@gmail.com> wrote:
I don’t think it’s possible to extend in a way
that you are asking (like, Java classes
“/extend/"). Though, you can create your own
composite PTransform that will incorporate one
or several others inside /“expand()”/ method.
Actually, most of the Beam native PTransforms
are composite transforms. Please, take a look on
doc and examples [1]
Regarding your example, please, be aware that
all PTransforms are supposed to be executed in
distributed environment and the order of records
is not guaranteed. So, limiting the whole output
by fixed number of records can be challenging -
you’d need to make sure that it will be
processed on only one worker, that means that
you’d need to shuffle all your records by the
same key and probably sort the records in way
that you need.
Did you consider to use
“/org.apache.beam.sdk.transforms.Top/” for that? [2]
If it doesn’t work for you, could you provide
more details of your use case? Then we probably
can propose the more suitable solutions for that.
[1]
https://beam.apache.org/documentation/programming-guide/#composite-transforms
[2]
https://beam.apache.org/releases/javadoc/2.50.0/org/apache/beam/sdk/transforms/Top.html
—
Alexey
On 15 Sep 2023, at 14:22, Joey Tran
<joey.t...@schrodinger.com> wrote:
Is there a way to extend already defined
PTransforms? My question is probably better
illustrated with an example. Let's say I have a
PTransform that generates a very variable
number of outputs. I'd like to "wrap" that
PTransform such that if it ever creates more
than say 1,000 outputs, then I just take the
first 1,000 outputs without generating the rest
of the outputs.
It'd be trivial if I have access to the DoFn,
but what if the PTransform in question doesn't
expose the `DoFn`?
