That is a really good way to describe my mental model as well.
On Tue, Jan 7, 2020 at 12:20 PM Kenneth Knowles <[email protected]
<mailto:[email protected]>> wrote:
On Tue, Jan 7, 2020 at 1:39 AM Jan Lukavský <[email protected]
<mailto:[email protected]>> wrote:
Hi Kenn,
I see that my terminology seems not to be 100% aligned
with Beam's. I'll work on that. :-)
I agree with what you say, and by "late" I mostly meant
"droppable" (arriving too late after watermark).
I'm definitely not proposing to get back to something
like "out of order" == "late" or anything like that. I'm
also aware that stateful operation is windowed operation,
but the semantics of the windowing is different than of a
GBK. The difference is how time moves in GBK and how
moves in stateful DoFn. Throwing away some details (early
triggers, late data triggers), the main difference is
that in GBK case, time hops just between window
boundaries, while in stateful DoFn time moves "smoothly"
(with each watermark update). Now, this difference brings
the question about why the definition of "droppable" data
is the same for both types of operations, when there is a
difference in how users "perceive" time. As the more
generic operation, stateful DoFn might deserve a more
general definition of droppable data, which should
degrade naturally to the one of GBK in presence of
"discrete time hops".
I understand what you mean. On the other hand, I encourage
thinking of event time spatially, not as time passing. That
is a big part of unifying batch/streaming real-time/archival
processing. The event time window is a secondary key to
partition the data (merging windows are slightly more
complex). All event time windows exist simultaneously. So for
both stateful ParDo and GBK, I find it helpful to consider
this perspective where all windows are processed
simultaneously / in an arbitrary order not assuming windows
are ordered at all. Then you see that GBK and stateful ParDo
do not really treat windows / watermark differently: both of
them process a stream of data for each (key, window) pair
until the watermark informs them that the stream is expired,
then they GC the state associated with that (key, window) pair.
Kenn
This might have some consequences on how the droppable
data should be handled in presence of (early) triggers,
because triggerring is actually what makes time to "hop",
so we might arrive to a conclusion that we might actually
drop any data that has timestamp less than "last trigger
time + allowed lateness". This looks appealing to me,
because IMO it has strong internal logical consistency.
Although it is possible that it would drop more data,
which is generally undesirable, I admit that.
I'm looking for explanation why the current approach was
chosen instead of the other.
Jan
On 1/7/20 12:52 AM, Kenneth Knowles wrote:
This thread has a lot in it, so I am just top-posting.
- Stateful DoFn is a windowed operation; state is
per-window. When the window expires, any further inputs
are dropped.
- "Late" is not synonymous with out-of-order. It
doesn't really have an independent meaning.
- For a GBK/Combine "late" means "not included prior
to the on-time output", and "droppable" means "arriving
after window expiry".
- For Stateful DoFn there is no real meaning to
"late" except if one is talking about "droppable", which
still means "arriving after window expiry". A user may
have a special timer where they flip a flag and treat
elements after the timer differently.
I think the definition of when data is droppable is very
simple. We explicitly moved to this definition, away
from the "out of order == late", because it is more
robust and simpler to think about. Users saw lots of
confusing behavior when we had "out of order by allowed
lateness == droppable" logic.
Kenn
On Mon, Jan 6, 2020 at 1:42 AM Jan Lukavský
<[email protected] <mailto:[email protected]>> wrote:
> Generally the watermark update can overtake
elements, because runners can explicitly ignore
late data in the watermark calculation (for good
reason - those elements are already late, so no need
to hold up the watermark advancing any more).
This seems not to affect the decision of _not late_
vs. _late_, is it? If element is late and gets
ignored from watermark calculation (whatever that
includes in this context), than the watermark cannot
move past elements that were not marked as _not
late_ and thus nothing can make them _late_.
> For GBK on-time data simply means the first pane
marked as on time. For state+timers I don't think it
makes sense for Beam to define on-time v.s. late,
rather I think the user can come up with their own
definition depending on their use case. For example,
if you are buffering data into BagState and setting
a timer to process it, it would be logical to say
that any element that was buffered before the timer
expired is on time, and any data that showed up
after the timer fired is late. This would roughly
correspond to what GBK does, and the answer would be
very similar to simply comparing against the
watermark (as the timers fire when the watermark
advances).
Yes, I'd say that stateful DoFns don't have (well
defined) concept of pane, because that is related to
concept of trigger and this is a concept of GBK (or
windowed operations in general). The only semantic
meaning of window in stateful DoFn is that it
"scopes" state.
This discussion might have got a little off the
original question, so I'll try to rephrase it:
Should stateful DoFn drop *all* late data, not just
data that arrive after window boundary + allowed
lateness? Some arguments why I think it should:
* in windowed operations (GBK), it is correct to
drop data on window boundaries only, because time
(as seen by user) effectively hops only on these
discrete time points
* in stateful dofn on the other hand time move
"smoothly" (yes, with some granularity, millisecond,
nanosecond, whatever and with watermark updates
only, but still)
* this could be viewed that dropping late data
immediately as time (again, from user perspective)
moves (not on some more or less artificial boundary
having only little semantic meaning) is consistent
with both the above properties
The negative side effect of this would be, that more
data could be dropped, but ... isn't this what
defines allowed lateness? I don't want to discuss
the implications on user pipelines of such a change
(and if we can or cannot do it), just trying to
build some theoretical understanding of the problem
as a whole. The decision if any change could /
should be made can be done afterwards.
Thanks,
Jan
On 1/4/20 10:35 PM, Reuven Lax wrote:
On Sat, Jan 4, 2020 at 12:13 PM Jan Lukavský
<[email protected] <mailto:[email protected]>> wrote:
> Yes, but invariants should hold. If I add a
ParDo that drops late elements (or, more
commonly,diverts the late elements to a
different PCollection), then the result of that
ParDo should _never_ introduce and more late
data. This cannot be guaranteed simply with
watermark checks. The ParDo may decide that the
element was not late, but by the time it
outputs the element the watermark may have
advanced, causing the element to actually be late.
This is actually very interesting. The question
is - if I decide about lateness based on output
watermark of a PTransform, is it still the
case, that in downstream operator(s) the
element could be changed from "not late" to
"late"? Provided the output watermark is
updated synchronously based on input data
(which should be) and watermark update cannot
"overtake" elements, I think that the
downstream decision should not be changed, so
the invariant should hold. Or am I missing
something?
Generally the watermark update can overtake
elements, because runners can explicitly ignore
late data in the watermark calculation (for good
reason - those elements are already late, so no
need to hold up the watermark advancing any more).
For GBK on-time data simply means the first pane
marked as on time. For state+timers I don't think
it makes sense for Beam to define on-time v.s.
late, rather I think the user can come up with
their own definition depending on their use case.
For example, if you are buffering data into
BagState and setting a timer to process it, it
would be logical to say that any element that was
buffered before the timer expired is on time, and
any data that showed up after the timer fired is
late. This would roughly correspond to what GBK
does, and the answer would be very similar to
simply comparing against the watermark (as the
timers fire when the watermark advances).
Reuven
On 1/4/20 8:11 PM, Reuven Lax wrote:
On Sat, Jan 4, 2020 at 11:03 AM Jan Lukavský
<[email protected] <mailto:[email protected]>> wrote:
On 1/4/20 6:14 PM, Reuven Lax wrote:
There is a very good reason not to define
lateness directly in terms of the
watermark. The model does not make any
guarantees that the watermark advances
synchronously, and in fact for the
Dataflow runner the watermark advances
asynchronously (i.e. independent of
element processing). This means that
simply comparing an element timestamp
against the watermark creates a race
condition. There are cases where the
answer could change depending on exactly
when you examine the watermark, and if
you examine again while processing the
same bundle you might come to a different
conclusion about lateness.
Due to monotonicity of watermark, I don't
think that the asynchronous updates of
watermark can change the answer from
"late" to "not late". That seems fine to me.
It's the other way around. You check to see
whether an element is late and the answer is
"not late." An instant later the answer
changes to "late" This does cause many
problems, and is why this was changed.
This non determinism is undesirable when
considering lateness, as it can break
many invariants that users may rely on
(e.g. if I could write a ParDo that
filtered all late data, yet still find
late data showing up downstream of the
ParDo which would be very surprising).
For that reason, the SDK always marks
things as late based on deterministic
signals. e.g. for a triggered GBK
everything in the first post-watermark
pane is marked as on time (no matter what
the watermark is) and everything in
subsequent panes is marked as late.
Dropping latecomers will always be
non-deterministic, that is certain. This
is true even in case where watermark is
updated synchronously with element
processing, due to shuffling and varying
(random) differences of processing and
event time in upstream operator(s). The
question was only if a latecomer should be
dropped only at a window boundaries only
(which is a sort of artificial time
boundary), or right away when spotted (in
stateful dofns only). Another question
would be if latecomers should be dropped
based on input or output watermark,
dropping based on output watermark seems
even to be stable in the sense, that all
downstream operators should come to the
same conclusion (this is a bit of a
speculation).
Yes, but invariants should hold. If I add a
ParDo that drops late elements (or, more
commonly,diverts the late elements to a
different PCollection), then the result of
that ParDo should _never_ introduce and more
late data. This cannot be guaranteed simply
with watermark checks. The ParDo may decide
that the element was not late, but by the time
it outputs the element the watermark may have
advanced, causing the element to actually be late.
In practice this is important. And early
version of Dataflow (pre Beam) implemented
lateness by comparing against the watermark,
and it caused no end of trouble for users.
FYI - this is also the reason why Beam
does not currently provide users direct
access to the watermark. The asynchronous
nature of it can be very confusing, and
often results in users writing bugs in
their pipelines. We decided instead to
expose easier-to-reason-about signals
such as timers (triggered by the
watermark), windows, and lateness.
Reuven
On Sat, Jan 4, 2020 at 1:15 AM Jan
Lukavský <[email protected]
<mailto:[email protected]>> wrote:
I realized the problem. I
misinterpreted the
LateDataDroppingDoFnRunner. It
doesn't drop *all* late (arriving
after watermark - allowed lateness)
data, but only data, that arrive
after maxTimestamp + allowedLateness
of their respective windows.
Stateful DoFn can run on global
window (which was the case of my
tests) and there is no dropping then.
Two questions arise then:
a) does it mean that this is one
more argument to move this logic to
StatefulDoFnRunner?
StatefulDoFnRunner performs state
cleanup on window GC time, so without
LateDataDroppingDoFnRunner and late
data will see empty state and will
produce wrong results.
b) is this behavior generally
intentional and correct? Windows and
triggers are (in my point of view)
features of GBK, not stateful DoFn.
Stateful DoFn is a low level
primitive, which can be viewed to
operate on "instant" windows, which
should then probably be defined as
dropping every single element arrive
after allowed lateness. This might
probably relate to question if
operations should be built bottom up
from most primitive and generic ones
to more specific ones - that is GBK
be implemented on top of stateful
DoFn and not vice versa.
Thoughts?
Jan
On 1/4/20 1:03 AM, Steve Niemitz wrote:
I do agree that the direct runner
doesn't drop late data arriving at a
stateful DoFn (I just tested as well).
However, I believe this is
consistent with other runners. I'm
fairly certain (at least last time I
checked) that at least Dataflow will
also only drop late data at GBK
operations, and NOT stateful DoFns.
Whether or not this is intentional
is debatable however, without being
able to inspect the watermark inside
the stateful DoFn, it'd be very
difficult to do anything useful with
late data.
On Fri, Jan 3, 2020 at 5:47 PM Jan
Lukavský <[email protected]
<mailto:[email protected]>> wrote:
I did write a test that tested
if data is dropped in a plain
stateful DoFn. I did this as
part of validating that PR [1]
didn't drop more data when using
@RequiresTimeSortedInput than it
would without this annotation.
This test failed and I didn't
commit it, yet.
The test was basically as follows:
- use TestStream to generate
three elements with timestamps
2, 1 and 0
- between elements with
timestamp 1 and 0 move watermark
to 1
- use allowed lateness of zero
- use stateful dofn that just
emits arbitrary data for each
input element
- use Count.globally to count
outputs
The outcome was that stateful
dofn using
@RequiresTimeSortedInput output
2 elements, without the
annotation it was 3 elements. I
think the correct one would be 2
elements in this case. The
difference is caused by the
annotation having (currently)
its own logic for dropping data,
which could be removed if we
agree, that the data should be
dropped in all cases.
On 1/3/20 11:23 PM, Kenneth
Knowles wrote:
Did you write such
a @Category(ValidatesRunner.class)
test? I believe the Java
direct runner does drop late
data, for both GBK and stateful
ParDo.
Stateful ParDo is implemented
on top of GBK:
https://github.com/apache/beam/blob/64262a61402fad67d9ad8a66eaf6322593d3b5dc/runners/direct-java/src/main/java/org/apache/beam/runners/direct/ParDoMultiOverrideFactory.java#L172
And GroupByKey, via
DirectGroupByKey, via
DirectGroupAlsoByWindow, does
drop late data:
https://github.com/apache/beam/blob/c2f0d282337f3ae0196a7717712396a5a41fdde1/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java#L220
I'm not sure why it has its own
code, since ReduceFnRunner also
drops late data, and it does
use ReduceFnRunner (the same
code path all Java-based
runners use).
Kenn
On Fri, Jan 3, 2020 at 1:02 PM
Jan Lukavský <[email protected]
<mailto:[email protected]>> wrote:
Yes, the non-reliability of
late data dropping in
distributed runner is
understood. But this is
even where DirectRunner can
play its role, because only
there it is actually
possible to emulate and
test specific watermark
conditions. Question
regarding this for the java
DirectRunner - should we
completely drop
LataDataDroppingDoFnRunner
and delegate the late data
dropping to
StatefulDoFnRunner? Seems
logical to me, as if we
agree that late data should
always be dropped, then
there would no "valid" use
of StatefulDoFnRunner
without the late data
dropping functionality.
On 1/3/20 9:32 PM, Robert
Bradshaw wrote:
I agree, in fact we just
recently enabled late data
dropping to the direct
runner in Python to be
able to develop better
tests for Dataflow.
It should be noted,
however, that in a
distributed runner (absent
the quiessence of
TestStream) that one can't
*count* on late data being
dropped at a certain
point, and in fact (due to
delays in fully
propagating the watermark)
late data can even become
on-time, so the promises
about what happens behind
the watermark are
necessarily a bit loose.
On Fri, Jan 3, 2020 at
9:15 AM Luke Cwik
<[email protected]
<mailto:[email protected]>>
wrote:
I agree that the
DirectRunner should
drop late data. Late
data dropping is
optional but the
DirectRunner is used
by many for testing
and we should have the
same behaviour they
would get on other
runners or users may
be surprised.
On Fri, Jan 3, 2020 at
3:33 AM Jan Lukavský
<[email protected]
<mailto:[email protected]>>
wrote:
Hi,
I just found out
that DirectRunner
is apparently not
using
LateDataDroppingDoFnRunner,
which means that
it doesn't drop
late data
in cases where
there is no GBK
operation involved
(dropping in GBK
seems
to be correct).
There is
apparently no
@Category(ValidatesRunner)
test
for that behavior
(because
DirectRunner would
fail it), so the
question
is - should late
data dropping be
considered part of
model (of which
DirectRunner
should be a
canonical
implementation)
and therefore that
should be fixed
there, or is the
late data dropping
an optional feature
of a runner?
I'm strongly in
favor of the first
option, and I
think it is likely
that
all real-world
runners would
probably adhere to
that (I didn't check
that, though).
Opinions?
Jan
