+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/
