+Mikhail Gryzykhin <mailto:[email protected]> +Rui Wang
<mailto:[email protected]> +Reza Rokni
<mailto:[email protected]> who have all done some
investigations here.
On Fri, Nov 22, 2019 at 11:48 AM Jan Lukavský
<[email protected] <mailto:[email protected]>> wrote:
On 11/22/19 7:54 PM, Reuven Lax wrote:
On Fri, Nov 22, 2019 at 10:19 AM Jan Lukavský
<[email protected] <mailto:[email protected]>> wrote:
Hi Reuven,
I didn't investigate that particular one, but
looking into that now, it looks that is (same as
the "classic" join library) builds around CoGBK. Is
that correct? If yes, then it essentially means
that it:
- works only for cases where both sides have the
same windowfn (that is limitation of Flatten that
precedes CoGBK)
Correct. Did you want to join different windows? If so
what are the semantics? If the lhs has FixedWindows and
the rhs has SessionWindows, what do you want the join
semantics to be? The only thing I could imagine would
be for the user to provide some function telling the
join how to map the windows together, but that could be
pretty complicated.
I don't want to go too far into details, but generally
both lhs and rhs can be put onto time line and then full
join can be defined as each pair of (lhs, first
preceding rhs) and (rhs, first preceding lhs). Then the
end of window is semantically just clearing the joined
value (setting it to null, thus at the end of window
there will be pair (lhs, null) or (null, rhs) in case of
full outer join). This way any combination of windows is
possible, because all window does is that it "scopes"
validity of respective values (lhs, rhs).
I think it is very valid to hope to do a join in the sense
of a relational join where it is row-to-row. In this case,
Beam's concept of windowing may or may not make sense. It is
just a tool for the job. It is just a grouping key that
provides a time when state can be deleted. So I would say
your use case is more global window to global window join.
That is what I think of as a true stream-to-stream join
anyhow. You probably don't want to wait forever for output.
So you'll need to use some knob other than Beam windows or
triggers.
Reza has prototyped a join like you describe here:
https://github.com/apache/beam/pull/9032
If your join condition explicitly includes the event time
distance between elements, then it could "just work". If
that isn't really part of your join condition, then you will
have to see this restriction as a "knob" that you tweak on
your results.
- when using global window, there has to be
trigger and (afaik) there is no trigger that would
guarantee firing after each data element (for early
panes) (because triggers are there to express
cost-latency tradeoff, not semantics)
Can you explain the use case where this matters? If you
do trigger elementCountAtLeast(1) on the join, then the
consumer will simply see a continuous stream of
outputs. I'm not sure I understand why the consumer
cares that some of those outputs were in a pane that
really held 3 outputs instead of 1.
What I'm trying to solve is basically this:
- lhs is event stream
- rhs is stream of a "state updates"
purpose of the join is "take each event, pair it with
currently valid state and produce output and possibly
modified state". I cannot process two events at a time,
because first event can modify the state and the
subsequent event should see this. It is not a "simple"
stateful pardo either, because the state can be modified
externally (not going into too much detail here, but
e.g. by writing into kafka topic).
Reuven's explanation is missing some detail. If the CoGBK is
in discarding mode, then it will miss join results. If the
CoGBK is in accumulating mode, it will duplicate join
results. This is a known problem and the general solution is
retractions.
Basically, CoGBK-based joins just don't work with triggers
until we have retractions.
Moreover, I'd like to define the join semantics so
that when there are available elements from both
sides, the fired pane should be ON_TIME, not EARLY.
That essentially means that the fully general case
would not be built around (Co)GBK, but stateful
ParDo. There are specific options where this fully
general case "degrades" into forms that can be
efficiently expressed using (Co)GBK, that is true.
BTW building this around stateful DoFn might be a
better fit. The main reason I didn't is because we
would need a good distributed MapState (something
discussed fairly recently on the list), and that is not
yet built. Once we had that, I might be inclined to
rewrite this join on stateful DoFn.
Yes, the sorted state helps for streaming case. But I'd
be careful about that for batch case, where this might
lead to high pressure on the state (and
InMemoryStateInternals might OOME for instance).
However can you explain what you are expecting from the
pane? An EARLY pane simply means that we are producing
output before the end of the window. If you are in the
global window triggering every element, then every
output is EARLY. It might seem weird if you are
interpreting EARLY as "outputting data that isn't
ready," however that's not what EARLY is defined to be.
Any change to the pane semantics would be a major
breaking change to very fundamental semantics.
I wonder if you are really objecting to the name EARLY
and ON_TIME? Maybe we would've been better off tagging
it BEFORE_WINDOW_END instead of EARLY, to make it clear
what is meant?
Essentially I don't object anything here. I'm missing
solution to the "event vs. state" join described above.
I was thinking about how to make these types of problems
more user friendly and it essentially leads to creating
a somewhat more generic semantics of join, where
end-of-window is converted into "'value-delete events"
and then just joining by the "previous" or "valid" value
(yes, this relates to validity windows mentioned on Beam
Summit Europe). It actually turns out that with some
work we could define quite "naturally" a join on two
streams with global window and no trigger. It would even
function with lowest latency possible (but yes, with the
highest expenses, it is actually the introduction of
(same!) windows that enable certain optimizations). It
the correctly defines semantics for different windows,
although the result would be (probably unexpectedly)
windowed using global window. But that doesn't seem to
be any breaking change, because it is currently not
possible (any such pipeline will not be validated).
Maybe for reference, the unwindowed join would be what
is described here [1]
[1]
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Join+Semantics#KafkaStreamsJoinSemantics-KStream-KTableJoin
Jan
On 11/22/19 6:47 PM, Reuven Lax wrote:
Have you seen the Join library that is part of
schemas? I'm curious whether this fits your needs,
or there's something lacking there.
On Fri, Nov 22, 2019 at 12:31 AM Jan Lukavský
<[email protected] <mailto:[email protected]>> wrote:
Hi,
based on roadmap [1], we would like to define
and implement a full set
of (unified) stream-stream joins. That would
include:
- joins (left, right, full outer) on global
window with "immediate
trigger"
- joins with different windowing functions
on left and right side
The approach would be to define these
operations in a natural way, so
that the definition is aligned with how
current joins work (same
windows, cartesian product of values with same
keys, output timestamp
projected to the end of window, etc.). Because
this should be a generic
approach, this effort should probably be part
of join library, that can
the be reused by other components, too (e.g. SQL).
The question is - is (or was) there any effort
that we can build upon?
Or should this be designed from scratch?
Jan
[1] https://beam.apache.org/roadmap/euphoria/
