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
