100% - the contract should not change because things are in
a bundle. IIRC there are some open bugs in Beam around this
that really should be fixed.
My issue with GroupIntoBatches is different. This transform
works as follows:
if (this is the first element in batch - checked by reading
a count stored in a ValueState)
timer.offset(bufferingDuration).setRelative()
This makes it tricky to use setTimer.withOutputTimestamp.
Inputs are not guaranteed to be in order, so simply adding a
withOutputTimestamp would set the timestamp to be whatever
the first element happened to be; it really should be the
minimum timestamp of all elements in the buffer. If we
started setting the timer on every element, then
timer.offset.setRelative would keep bumping the
(processing-time) timer into the future and it would never
expire.
One solution would be to store the timer timestamp in a
ValueState, and use Timer.set to set the timer to an
absolute timestamp. This would allow us to always reset the
timer to the same expiration target, just modifying the
output timestamp each time. However, this will break
DirectRunner tests. The DirectRunner allows the user to
control the advancement of processing time when using
TestStream, but this facility doesn't work well if the
transform sets the processing-time timer using absolute
set() calls.
I'm not sure how to solve this using the existing Timer API.
On Wed, May 19, 2021 at 4:39 PM Robert Bradshaw
<rober...@google.com <mailto:rober...@google.com>> wrote:
+1. It was my understanding as well that consensus was
that timers
must be delivered in timestamp order, and "within bundle"
resetting/clearing of timers should be respected (as if
each timer was
in its own bundle).
On Wed, May 19, 2021 at 3:01 PM Kenneth Knowles
<k...@apache.org <mailto:k...@apache.org>> wrote:
>
> Reading over the other thread, there was consensus to
implement.
>
> Reading commentary on the PR, there were good
questions raised about the semantics. Questions which I
feel able to have an opinion about :-)
>
> The questions surrounded bundling and timers in the
same bundling clearing each other. Actually the same
questions apply to timers re-setting later timers and
+Jan Lukavský has raised this already (among other
people) so we kind of know the answer now, and I think
+Boyuan Zhang code was good (from my quick read). What
has changed is that we have a better idea of the
contract with the runner. I'm not sure if portability
makes this more complex. I will share all my thoughts on
this:
>
> I think one key to the Beam model is that bundles are
for performance and also intended as the unit of
commitment (until FinishBundle is called, there may be
unfinished work). They can affect *behavior* (what the
program does - including what can be observed) ... but
not *semantics* (what the output means).
>
> So, for example, bundling affects how many files are
written but you are expected to read all the files and
the number or ordering you must not depend on. The
behavior is different, but the semantics are the same.
>
> When it comes to timers, behavior and semantics are
very tightly coupled; timers are like a self loop. The
firing of a timer is a behavior w.r.t. the output of the
stateful ParDo but it is semantics of the timer itself
(my definitions don't actually apply so simply so don't
dig too hard :-). So to get bundling-invariant
semantics, we should try for bundling-invariant
behavior. When some clock moves from T to T+D between
bundles then all the timers in the range [T, T+D) may
fire so they are delivered in the bundle. I believe we
have in the prior re-setting issue agreed that timers
should always be called back in timestamp order.
Bundling-invariance then implies that earlier timers can
clear later timers just like they can re-set them. So
even though a timer is delivered in a bundle, the local
state of the timer wins. This matches how state works as
well; no matter how things are bundled, the state you
read is always whatever was written last.
>
> Kenn
>
> On Tue, May 11, 2021 at 1:24 PM Siyuan Chen
<syc...@google.com <mailto:syc...@google.com>> wrote:
>>
>>
>>
>> On Tue, May 11, 2021 at 11:08 AM Reuven Lax
<re...@google.com <mailto:re...@google.com>> wrote:
>>>
>>>
>>>
>>> On Tue, May 11, 2021 at 9:01 AM Kenneth Knowles
<k...@apache.org <mailto:k...@apache.org>> wrote:
>>>>
>>>>
>>>>
>>>> On Mon, May 10, 2021 at 7:40 PM Reuven Lax
<re...@google.com <mailto:re...@google.com>> wrote:
>>>>>
>>>>> Hi,
>>>>>
>>>>> I've been looking at the implementation of
GroupIntoBatches (hoping to add support to group based
on byte size), and I have a few questions about the
current implementation.
>>>>>
>>>>> 1. I noticed that the transform does not preserve
input timestamps. The output timestamp is not explicitly
set, so it will be whatever the default output
timestamp. Confusingly this will depend on which path is
taken. If the batch is output while processing an
element, then the timestamp of tha element will be used.
If the batch is output from the timer, then the
processing-time value of the timer will be used.
>>>>
>>>>
>>>> Seems like bugs.
>>>>
>>>>>
>>>>> - Should I start setting outputTimestamp
explicitly - ideally to the minimum timestamp in the
current batch of elements?
>>>>
>>>>
>>>> That's a sensible default. Could pass a
TimestampCombiner if that leads to watermark delays.
Given the nature of the transform, the simple approach
will probably be fine.
>>>
>>>
>>> This is a bit tricky to do. in order to update the
output timestamp, we need to call setTimer again.
However since we are calling
timer.offset().setRelative(), this will keep bumping the
timer into the future and it will never fire.
>>>
>>> One solution would be for GIB to just store the
current timer ts in state, and make sure that keep
setting the same target until the timer fires. However
that precludes us from using setRelative (since that
method does not allow the DoFn to see what target time
is being set). I think that this approach won't play
nice with TestStream.advanceProcessingTimeTo.
>>>
>>> We could also add a way to update _just_ the output
time for a timer without resetting the entire timer.
>>>
>>>>
>>>>
>>>>> - Should we have the option to preserve all
element timestamps? this could look something like this:
>>>>>
>>>>> PColllection<KV<K, TimestampedValue<V>>> batches =
>>>>>
input.apply(GroupIntoBatches.ofSize(N).withTimestampedValues());
>>>>
>>>>
>>>> This seems useful.
>>>>
>>>>>
>>>>> 2. flushBatch always resets the timer, even after
the batch is processed. The only reason I can think of
for doing so is to update the watermark hold.
TimerInternals contains a deleteTimer method - is there
any reason we shouldn't simply implement Timer.clear and
hook it up to the existing deleteTimer?
>>>>
>>>>
>>>> Makes sense to me. Prior thread on this seemed to
have lightweight consensus:
https://lists.apache.org/thread.html/r91af7dff0070b80b275082ca0cff7486dc5dfdfc113f35f560600792%40%3Cdev.beam.apache.org%3E
<https://lists.apache.org/thread.html/r91af7dff0070b80b275082ca0cff7486dc5dfdfc113f35f560600792%40%3Cdev.beam.apache.org%3E>
>>
>> There was another thread about adding Timer.clear in
Java:
https://lists.apache.org/thread.html/r47f5f14ba9729a2800222440425543ef40bea12eef1f2739d42e75db%40%3Cdev.beam.apache.org%3E
<https://lists.apache.org/thread.html/r47f5f14ba9729a2800222440425543ef40bea12eef1f2739d42e75db%40%3Cdev.beam.apache.org%3E>
>> and a previous attempt in
https://github.com/apache/beam/pull/12836
<https://github.com/apache/beam/pull/12836>. Boyuan
might have more insights into this +Boyuan Zhang
>>>>
>>>>
>>>>>
>>>>> 3. This transform was implemented before
OnWindowExpiration was implemented. I think we should
add a window-expiration callback, and stop setting the
end-of window timer.
>>>>
>>>>
>>>> +1
>>>>
>>>> Kenn
