On Mon, Oct 24, 2022 at 5:51 AM Jan Lukavský <je...@seznam.cz> wrote:
> On 10/22/22 21:47, Reuven Lax via dev wrote: > > I think we stated that CoGroupbyKey was also a primitive, though in > practice it's implemented in terms of GroupByKey today. > > On Fri, Oct 21, 2022 at 3:05 PM Kenneth Knowles <k...@apache.org> wrote: > >> >> >> On Fri, Oct 21, 2022 at 5:24 AM Jan Lukavský <je...@seznam.cz> wrote: >> >>> Hi, >>> >>> I have some missing pieces in my understanding of the set of Beam's >>> primitive transforms, which I'd like to fill. First a quick recap of what I >>> think is the current state. We have (basically) the following primitive >>> transforms: >>> >>> - DoFn (stateless, stateful, splittable) >>> >>> - Window >>> >>> - Impulse >>> >>> - GroupByKey >>> >>> - Combine >>> >> >> Not a primitive, just a well-defined transform that runners can execute >> in special ways. >> > Yep, OK, agree. Performance is orthogonal to semantics. > > >> >>> >>> >>> - Flatten (pCollections) >>> >> >> The rest, yes. >> >> >> >>> Inside runners, we most often transform GBK into ReduceFn >>> (ReduceFnRunner), which does the actual logic for both GBK and stateful >>> DoFn. >>> >> >> ReduceFnRunner is for windowing / triggers and has special feature to use >> a CombineFn while doing it. Nothing to do with stateful DoFn. >> > My bad, wrong wording. The point was that *all* of the semantics of GBK > and Combine can be defined in terms of stateful DoFn. There are some > changes needed to stateful DoFn to support the Combine functionality. But > as mentioned above - optimization is orthogonal to semantics. > Not quite IMO. It is a subtle difference. Perhaps these transforms can be *implemented* using stateful DoFn, but defining their semantics directly at a high level is more powerful. The higher level we can make transforms, the more flexibility we have in the runners. You *could* suggest that we take the same approach as we do with Combine: not a primitive, but a special transform that we optimize. You could say that "vanilla ParDo" is a composite that has a stateful ParDo implementation, but a runner can implement the composite more efficiently (without a shuffle). Same with CoGBK. You could say that there is a default expansion of CoGBK that uses stateful DoFn (which implies a shuffle) but that smart runners will not use that expansion. > >> >> >>> I'll compare this to the set of transforms we used to use in Euphoria >>> (currently java SDK extension): >>> >>> - FlatMap ~~ stateless DoFn >>> >>> - Union ~~ Flatten >>> >>> - ReduceStateByKey ~~ stateful DoFn, GBK, Combine, Window >>> >> >> Stateful DoFn does not require associative or commutative operation, >> while reduce/combine does. Windowing is really just a secondary key for >> GBK/Combine that allows completion of unbounded aggregations but has no >> computation associated with it. >> > Merging WindowFn contains some computation. The fact that stateful DoFn do > not require specific form of reduce function is precisely what makes it the > actual primitive, no? > > >> >> >>> - (missing Impulse) >>> >> >> Then you must have some primitive sources with splitting? >> >> >>> - (missing splittable DoFn) >>> >> >> Kind of the same question - SDF is the one and only primitive that >> creates parallelism. >> > Original Euphoria had an analogy to (Un)boundedReader. The SDK extension > in Beam works on top of PCollecions and therefore does not deal with IOs. > > >> The ReduceStateByKey is a transform that is a "combinable stateful DoFn" >>> - i.e. the state might be created pre-shuffle, on trigger the state is >>> shuffled and then merged. In Beam we already have CombiningState and >>> MergingState facility (sort of), which is what is needed, we just do not >>> have the ability to shuffle the partial states and then combine them. This >>> also relates to the inability to run stateful DoFn for merging windowFns, >>> because that is needed there as well. Is this something that is >>> fundamentally impossible to define for all runners? What is worth noting is >>> that building, shuffling and merging the state before shuffle requires >>> compatible trigger (purely based on watermark), otherwise the transform >>> fall-backs to "classical DoFn". >>> >> >> Stateful DoFn for merging windows can be defined. You could require all >> state to be mergeable and then it is automatic. Or you could have an >> "onMerge" callback. These should both be fine. The automatic version is >> less likely to have nonsensical semantics, but allowing the callback to do >> "whatever it wants" whether the result is good or not is more consistent >> with the design of stateful DoFn. >> > Yes, but this is the same for CombineFn, right? The merge (or combine) has > to be correctly aligned with the computation. The current situation is that > we do not support stateful DoFns for merging WindowFn [1]. > > >> Whether and where a shuffle takes place may vary. Start with the maths. >> > Shuffle happens at least whenever there is a need to regroup keys. I'm not > sure which maths you refer to, can you clarify please? > > Jan > > [1] > https://github.com/apache/beam/blob/45b6ac71a87bb2ed83613c90d35ef2d0752266bf/runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java#L106 > > >> Kenn >> >> >>> Bottom line: I'm thinking of proposing to drop Euphoria extension, >>> because it has essentially no users and actually no maintainers, but I have >>> a feeling there is a value in the set of operators that could be >>> transferred to Beam core, maybe. I'm pretty sure it would bring value to >>> users to have access to a "combining stateful DoFn" primitive (even better >>> would be "combining splittable DoFn"). >>> >>> Looking forward to any comments on this. >>> >>> Jan >>> >>> >>>
