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
