Yea I like DelayTimer, or SleepTimer, or WaitTimer or some such.
OutputTime is always an event time timestamp so it isn't even
allowed to be set outside the window (or you'd end up with an
element assigned to a window that it isn't within, since
OutputTime essentially represents reserving the right to output
an element with that timestamp)
Kenn
On Mon, Feb 26, 2024 at 3:19 PM Robert Burke <rob...@frantil.com>
wrote:
Agreed that a retroactive behavior change would be bad, even
if tied to a beam version change. I agree that it meshes well
with the general theme of State & Timers exposing underlying
primitives for implementing Windowing and similar. I'd say
the distinction between the two might be additional
complexity for users to grok, and would need to be documented
well, as both operate in the ProcessingTime domain, but
differently.
What to call this new timer then? DelayTimer?
"A DelayTimer sets an instant in ProcessingTime at which
point computations can continue. Runners will prevent the
EventTimer watermark from advancing past the set OutputTime
until Processing Time has advanced to at least the provided
instant to execute the timers callback. This can be used to
allow the runner to constrain pipeline throughput with user
guidance."
I'd probably add that a timer with an output time outside of
the window would not be guaranteed to fire, and that
OnWindowExpiry is the correct way to ensure cleanup occurs.
No solution to the Looping Timers on Drain problem here, but
i think that's ultimately an orthogonal discussion, and will
restrain my thoughts on that for now.
This isn't a proposal, but exploring the solution space
within our problem. We'd want to break down exactly what
different and the same for the 3 kinds of timers...
On Mon, Feb 26, 2024, 11:45 AM Kenneth Knowles
<k...@apache.org> wrote:
Pulling out focus points:
On Fri, Feb 23, 2024 at 7:21 PM Robert Bradshaw via dev
<dev@beam.apache.org> wrote:
> I can't act on something yet [...] but I expect to be
able to [...] at some time in the processing-time future.
I like this as a clear and internally-consistent feature
description. It describes ProcessContinuation and those
timers which serve the same purpose as ProcessContinuation.
On Fri, Feb 23, 2024 at 7:21 PM Robert Bradshaw via dev
<dev@beam.apache.org> wrote:
> I can't think of a batch or streaming scenario where it
would be correct to not wait at least that long
The main reason we created timers: to take action in the
absence of data. The archetypal use case for processing
time timers was/is "flush data from state if it has been
sitting there too long". For this use case, the right
behavior for batch is to skip the timer. It is actually
basically incorrect to wait.
On Fri, Feb 23, 2024 at 3:54 PM Robert Burke
<lostl...@apache.org> wrote:
> It doesn't require a new primitive.
IMO what's being proposed *is* a new primitive. I think
it is a good primitive. It is the underlying primitive to
ProcessContinuation. It would be user-friendly as a kind
of timer. But if we made this the behavior of processing
time timers retroactively, it would break everyone using
them to flush data who is also reprocessing data.
There's two very different use cases ("I need to wait,
and block data" vs "I want to act without data, aka NOT
wait for data") and I think we should serve both of them,
but it doesn't have to be with the same low-level feature.
Kenn
On Fri, Feb 23, 2024 at 7:21 PM Robert Bradshaw via dev
<dev@beam.apache.org> wrote:
On Fri, Feb 23, 2024 at 3:54 PM Robert Burke
<lostl...@apache.org> wrote:
>
> While I'm currently on the other side of the fence,
I would not be against changing/requiring the
semantics of ProcessingTime constructs to be "must
wait and execute" as such a solution, and enables the
Proposed "batch" process continuation throttling
mechanism to work as hypothesized for both "batch"
and "streaming" execution.
>
> There's a lot to like, as it leans Beam further
into the unification of Batch and Stream, with one
fewer exception (eg. unifies timer experience
further). It doesn't require a new primitive. It
probably matches more with user expectations anyway.
>
> It does cause looping timer execution with
processing time to be a problem for Drains however.
I think we have a problem with looping timers plus
drain (a mostly
streaming idea anyway) regardless.
> I'd argue though that in the case of a drain, we
could updated the semantics as "move watermark to
infinity" "existing timers are executed, but new
timers are ignored",
I don't like the idea of dropping timers for drain. I
think correct
handling here requires user visibility into whether a
pipeline is
draining or not.
> and ensure/and update the requirements around
OnWindowExpiration callbacks to be a bit more
insistent on being implemented for correct execution,
which is currently the only "hard" signal to the SDK
side that the window's work is guaranteed to be over,
and remaining state needs to be addressed by the
transform or be garbage collected. This remains
critical for developing a good pattern for
ProcessingTime timers within a Global Window too.
+1
>
> On 2024/02/23 19:48:22 Robert Bradshaw via dev wrote:
> > Thanks for bringing this up.
> >
> > My position is that both batch and streaming
should wait for
> > processing time timers, according to local time
(with the exception of
> > tests that can accelerate this via faked clocks).
> >
> > Both ProcessContinuations delays and
ProcessingTimeTimers are IMHO
> > isomorphic, and can be implemented in terms of
each other (at least in
> > one direction, and likely the other). Both are an
indication that I
> > can't act on something yet due to external
constraints (e.g. not all
> > the data has been published, or I lack sufficient
capacity/quota to
> > push things downstream) but I expect to be able
to (or at least would
> > like to check again) at some time in the
processing-time future. I
> > can't think of a batch or streaming scenario
where it would be correct
> > to not wait at least that long (even in batch
inputs, e.g. suppose I'm
> > tailing logs and was eagerly started before they
were fully written,
> > or waiting for some kind of (non-data-dependent)
quiessence or other
> > operation to finish).
> >
> >
> > On Fri, Feb 23, 2024 at 12:36 AM Jan Lukavský
<je...@seznam.cz> wrote:
> > >
> > > For me it always helps to seek analogy in our
physical reality. Stream
> > > processing actually has quite a good analogy
for both event-time and
> > > processing-time - the simplest model for this
being relativity theory.
> > > Event-time is the time at which events occur
_at distant locations_. Due
> > > to finite and invariant speed of light (which
is actually really
> > > involved in the explanation why any stream
processing is inevitably
> > > unordered) these events are observed
(processed) at different times
> > > (processing time, different for different
observers). It is perfectly
> > > possible for an observer to observe events at a
rate that is higher than
> > > one second per second. This also happens in
reality for observers that
> > > travel at relativistic speeds (which might be
an analogy for fast -
> > > batch - (re)processing). Besides the invariant
speed, there is also
> > > another invariant - local clock (wall time)
always ticks exactly at the
> > > rate of one second per second, no matter what.
It is not possible to
> > > "move faster or slower" through (local) time.
> > >
> > > In my understanding the reason why we do not
put any guarantees or
> > > bounds on the delay of firing processing time
timers is purely technical
> > > - the processing is (per key) single-threaded,
thus any timer has to
> > > wait before any element processing finishes.
This is only consequence of
> > > a technical solution, not something fundamental.
> > >
> > > Having said that, my point is that according to
the above analogy, it
> > > should be perfectly fine to fire processing
time timers in batch based
> > > on (local wall) time only. There should be no
way of manipulating this
> > > local time (excluding tests). Watermarks should
be affected the same way
> > > as any buffering in a state that would happen
in a stateful DoFn (i.e.
> > > set timer holds output watermark). We should
probably pay attention to
> > > looping timers, but it seems possible to define
a valid stopping
> > > condition (input watermark at infinity).
> > >
> > > Jan
> > >
> > > On 2/22/24 19:50, Kenneth Knowles wrote:
> > > > Forking this thread.
> > > >
> > > > The state of processing time timers in this
mode of processing is not
> > > > satisfactory and is discussed a lot but we
should make everything
> > > > explicit.
> > > >
> > > > Currently, a state and timer DoFn has a
number of logical watermarks:
> > > > (apologies for fixed width not coming through
in email lists). Treat
> > > > timers as a back edge.
> > > >
> > > > input --(A)----(C)--> ParDo(DoFn) ----(D)--->
output
> > > > ^ |
> > > > |--(B)-----------------|
> > > > timers
> > > >
> > > > (A) Input Element watermark: this is the
watermark that promises there
> > > > is no incoming element with a timestamp
earlier than it. Each input
> > > > element's timestamp holds this watermark.
Note that *event time timers
> > > > firing is according to this watermark*. But a
runner commits changes
> > > > to this watermark *whenever it wants*, in a
way that can be
> > > > consistent. So the runner can absolute
process *all* the elements
> > > > before advancing the watermark (A), and only
afterwards start firing
> > > > timers.
> > > >
> > > > (B) Timer watermark: this is a watermark that
promises no timer is set
> > > > with an output timestamp earlier than it.
Each timer that has an
> > > > output timestamp holds this watermark. Note
that timers can set new
> > > > timers, indefinitely, so this may never reach
infinity even in a drain
> > > > scenario.
> > > >
> > > > (C) (derived) total input watermark: this is
a watermark that is the
> > > > minimum of the two above, and ensures that
all state for the DoFn for
> > > > expired windows can be GCd after calling
@OnWindowExpiration.
> > > >
> > > > (D) output watermark: this is a promise that
the DoFn will not output
> > > > earlier than the watermark. It is held by the
total input watermark.
> > > >
> > > > So a any timer, processing or not, holds the
total input watermark
> > > > which prevents window GC, hence the timer
must be fired. You can set
> > > > timers without a timestamp and they will not
hold (B) hence not hold
> > > > the total input / GC watermark (C). Then if a
timer fires for an
> > > > expired window, it is ignored. But in general
a timer that sets an
> > > > output timestamp is saying that it may
produce output, so it *must* be
> > > > fired, even in batch, for data integrity.
There was a time before
> > > > timers had output timestamps that we said
that you *always* have to
> > > > have an @OnWindowExpiration callback for data
integrity, and
> > > > processing time timers could not hold the
watermark. That is changed now.
> > > >
> > > > One main purpose of processing time timers in
streaming is to be a
> > > > "timeout" for data buffered in state, to
eventually flush. In this
> > > > case the output timestamp should be the
minimum of the elements in
> > > > state (or equivalent). In batch, of course,
this kind of timer is not
> > > > relevant and we should definitely not wait
for it, because the goal is
> > > > to just get through all the data. We can
justify this by saying that
> > > > the worker really has no business having any
idea what time it really
> > > > is, and the runner can just run the clock at
whatever speed it wants.
> > > >
> > > > Another purpose, brought up on the Throttle
thread, is to wait or
> > > > backoff. In this case it would be desired for
the timer to actually
> > > > cause batch processing to pause and wait.
This kind of behavior has
> > > > not been explored much. Notably the runner
can absolutely process all
> > > > elements first, then start to fire any
enqueued processing time
> > > > timers. In the same way that state in batch
can just be in memory,
> > > > this *could* just be a call to sleep(). It
all seems a bit sketchy so
> > > > I'd love clearer opinions.
> > > >
> > > > These two are both operational effects - as
you would expect for
> > > > processing time timers - and they seem to be
in conflict. Maybe they
> > > > just need different features?
> > > >
> > > > I'd love to hear some more uses of processing
time timers from the
> > > > community.
> > > >
> > > > Kenn
> >
