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
