Creating composite DoFns is tricky today due to how they are implemented (via annotated methods). However providing such a method to compose DoFns would be very useful IMO.
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`? >>>> >>>> >>>>
