This is a bit of a digression, but at least for the Dataflow runner the
only way to implement that would be to add an extra ValueState per timer
(which would add overhead). Timers were implemented based around blind
writes, and weren't designed for point reads. However in some cases the
ability to have dynamic timers actually solves use cases which otherwise
would need this functionality.
On Tue, Oct 22, 2019 at 4:56 PM Reza Rokni <r...@google.com> wrote:
> +1 on this, having the ability to create timers based on data would make a
> bunch of use cases easier to write.
>
> Any thoughts on having a isSet() / read() / setMinimum(timeStamp) type
> ability?
>
> On Wed, 23 Oct 2019 at 00:52, Reuven Lax <re...@google.com> wrote:
>
>> Kenn:
>> +1 to using TimerFamily instead of TimerId and TimerMap.
>>
>> Jan:
>> This is definitely not just for DSLs. I've definitely seen cases where
>> the user wants different timers based on input data, so they cannot be
>> defined statically. As a thought experiment: one stated goal of state +
>> timers was to provide the low-level tools we use to implement windowing.
>> However to implement windowing you need a dynamic set of timers, not just a
>> single one. Now most users don't need to reimplement windowing (though we
>> have had some users who had that need, when they wanted something slightly
>> different than what native Beam windowing provided), however the need for
>> dynamic timers is not unheard of.
>>
>> +1 to allowing dynamic state. However I think this is separate enough
>> from timers that it doesn't need to be coupled in this discussion. Dynamic
>> state also raises the wrinkle of pipeline compatibility (as you mentioned),
>> which I think is a bit less of an issue for dynamic timers.
>>
>> Allowing a DSL to specify a DoFnSignature does not quite solve this
>> problem. The DSL still needs a way to set and process the timers. It also
>> does not solve the problem where the timers are based on input data
>> elements, so cannot be known at pipeline construction time. However what
>> might be more important is statically defining the timer families, and a
>> DSL could do this by specifying a DoFnSignature (and something similar
>> could be done with state). Also as mentioned above, this is useful to
>> normal Beam users as well, and we shouldn't force normal users to start
>> dealing with DoFnSignatures and DoFnInvokers.
>>
>>
>>
>>
>>
>>
>> On Tue, Oct 22, 2019 at 7:56 AM Jan Lukavský <je...@seznam.cz> wrote:
>>
>>> Hi Max,
>>>
>>> wouldn't that be actually the same as
>>>
>>> class MyDoFn extends DoFn<String, String> {
>>>
>>>
>>> @ProcessElement
>>> public void process(
>>> ProcessContext context) {
>>> // "get" would register a new TimerSpec
>>> Timer timer1 = context.getTimer("timer1");
>>> Timer timer2 = context.getTimer("timer2");
>>> timers.set(...);
>>> timers.set(...);
>>> }
>>>
>>> That is - no need to declare anything? One more concern about that - if
>>> we allow registration of timers (or even state) dynamically like that it
>>> might be harder to perform validation of pipeline upon upgrades.
>>>
>>> Jan
>>>
>>> On 10/22/19 4:47 PM, Maximilian Michels wrote:
>>> > The idea makes sense to me. I really like that Beam gives upfront
>>> > specs for timer and state, but it is not flexible enough for
>>> > timer-based libraries or for users which want to dynamically generate
>>> > timers.
>>> >
>>> > I'm not sure about the proposed API yet. Shouldn't we separate the
>>> > timer specs from setting actual timers?
>>> >
>>> > Suggestion:
>>> >
>>> > class MyDoFn extends DoFn<String, String> {
>>> > @TimerMap TimerMap timers = TimerSpecs.timerMap();
>>> >
>>> > @ProcessElement
>>> > public void process(
>>> > @Element String e,
>>> > @TimerMap TimerMap timers)) {
>>> > // "get" would register a new TimerSpec
>>> > Timer timer1 = timers.get("timer1");
>>> > Timer timer2 = timers.get("timer2");
>>> > timers.set(...);
>>> > timers.set(...);
>>> > }
>>> >
>>> > // No args for "@OnTimer" => use generic TimerMap
>>> > @OnTimer
>>> > public void onTimer(
>>> > @TimerId String timerFired,
>>> > @Timestamp Instant timerTs,
>>> > @TimerMap TimerMap timers) {
>>> > // Timer firing
>>> > ...
>>> > // Set this timer (or another)
>>> > Timer timer = timers.get(timerFired);
>>> > timer.set(...);
>>> > }
>>> > }
>>> >
>>> > What do you think?
>>> >
>>> > -Max
>>> >
>>> > On 22.10.19 10:35, Jan Lukavský wrote:
>>> >> Hi Kenn,
>>> >>
>>> >> On 10/22/19 2:48 AM, Kenneth Knowles wrote:
>>> >>> This seems extremely useful.
>>> >>>
>>> >>> I assume you mean `@OnTimer("timers")` in your example. I would
>>> >>> suggest that the parameter annotation be something other
>>> >>> than @TimerId since that annotation is already used for a very
>>> >>> similar but different purpose; they are close enough that it is
>>> >>> tempting to pun them, but it is clearer to keep them distinct IMO.
>>> >>> Perhaps @TimerName or @TimerKey or some such. Alternatively,
>>> >>> keep @TimerId in the parameter list and change the declaration
>>> >>> to @TimerFamily("timers"). I think "family" or "group" may be more
>>> >>> clear naming than "map".
>>> >>>
>>> >>> At the portability level, this API does seem to be pretty close to a
>>> >>> noop in terms of the messages that needs to be sent over the Fn API,
>>> >>> so it makes sense to loosen the protos. By the time the Fn API is in
>>> >>> play, all of our desires to catch errors prior to execution are
>>> >>> irrelevant anyhow.
>>> >>>
>>> >>> On the other hand, I think DSLs have a different & bigger problem
>>> >>> than this, in that they want to programmatically adjust all the
>>> >>> capabilities of a DoFn. Same goes for wrapping one DoFn in
>>> >>> another. Certainly some limited DSL use cases are addressed by this,
>>> >>> but I wouldn't take that as a primary use case for this feature.
>>> >>> Ultimately they are probably better served by being able to
>>> >>> explicitly author a DoFnInvoker and provide it to a variant of
>>> >>> beam:transforms:ParDo where the do_fn field is a serialized
>>> >>> DoFnInvoker. Now that I think about this, I cannot recall why we
>>> >>> don't already ship a DoFnSignature & DoFnInvoker as the payload.
>>> >>> That would allow maximum flexibility in utilizing the portability
>>> >>> framework.
>>> >>
>>> >> yes, exactly, but when DSLs are in question, we have to make sure
>>> >> that DSLs are not bound to portability - we have to be able to
>>> >> translate even in case of "legacy" runners as well. That might
>>> >> complicate things a bit maybe.
>>> >>
>>> >> Jan
>>> >>
>>> >>>
>>> >>> Kenn
>>> >>>
>>> >>> On Mon, Oct 21, 2019 at 3:23 PM Reuven Lax <re...@google.com
>>> >>> <mailto:re...@google.com>> wrote:
>>> >>>
>>> >>> BEAM-6857 documents the need for dynamic timer support in the
>>> Beam
>>> >>> API. I wanted to make a proposal for what this API would look
>>> >>> like, and how to express it in the portability protos.
>>> >>>
>>> >>> Background: Today Beam (especially BeamJava) requires a ParDo to
>>> >>> statically declare all timers it accesses at compile time. For
>>> >>> example:
>>> >>>
>>> >>> class MyDoFn extends DoFn<String, String> {
>>> >>> @TimerId("timer1") TimerSpec timer1 =
>>> >>> TimerSpecs.timer(TimeDomain(EVENT_TIME));
>>> >>> @TimerId("timer2") TimerSpec timer2 =
>>> >>> TimerSpecs.timer(TimeDomain(PROCESSING_TIME));
>>> >>>
>>> >>> @ProcessElement
>>> >>> public void process(@Element String e, @TimerId("timer1") Timer
>>> >>> timer1, @TimerId("timer2") Timer timer2)) {
>>> >>> timer1.set(...);
>>> >>> timer2.set(...);
>>> >>> }
>>> >>>
>>> >>> @OnTimer("timer1") public void onTimer1() { ... }
>>> >>> @OnTimer("timer2") public void onTimer2() { ... }
>>> >>> }
>>> >>>
>>> >>> This requires the author of a ParDo to know the full list of
>>> >>> timers ahead of time, which has been problematic in many cases.
>>> >>> One example where it causes issues is for DSLs such as Euphoria
>>> or
>>> >>> Scio. DSL authors usually write ParDos to interpret the code
>>> >>> written in the high-level DSL, and so don't know ahead of time
>>> the
>>> >>> list of timers needed; alternatives today are quite ugly:
>>> physical
>>> >>> code generation or creating a single timer that multiplexes all
>>> of
>>> >>> the users logical timers. There are also cases where a ParDo
>>> needs
>>> >>> multiple distinct timers, but the set of distinct timers is
>>> >>> controlled by the input data, and therefore not knowable in
>>> >>> advance. The Beam timer API has been insufficient for these use
>>> >>> cases.
>>> >>>
>>> >>> I propose a new TimerMap construct, which allow a ParDo to
>>> >>> dynamically set named timers. It's use in the Java API would look
>>> >>> as follows:
>>> >>>
>>> >>> class MyDoFn extends DoFn<String, String> {
>>> >>> @TimerId("timers") TimerSpec timers =
>>> >>> TimerSpecs.timerMap(TimeDomain(EVENT_TIME));
>>> >>>
>>> >>> @ProcessElement
>>> >>> public void process(@Element String e, @TimerId("timers")
>>> >>> TimerMap timer)) {
>>> >>> timers.set("timer1", ...);
>>> >>> timers.set("timer2", ...);
>>> >>> }
>>> >>>
>>> >>> @OnTimer("timer") public void onTimer(@TimerId String
>>> >>> timerFired, @Timestamp Instant timerTs) { ... }
>>> >>> }
>>> >>>
>>> >>> There is a new TimerSpec type to specify a TimerMap. The TimerMap
>>> >>> class itself allows dynamically setting multiple timers based on
>>> a
>>> >>> String tag argument. Each TimerMap has a single callback which
>>> >>> when called is given the id of the timer that is currently
>>> firing.
>>> >>>
>>> >>> It is allowed to have multiple TimerMap objects in a ParDo (and
>>> >>> required if you want to have both processing-time and event-time
>>> >>> timers in the same ParDo). Each TimerMap is its own logical
>>> >>> namespace. i.e. if the user sets timers with the same string tag
>>> >>> on different TimerMap objects the timers will not collide.
>>> >>>
>>> >>> Currently the portability protos were written to mirror the Java
>>> >>> API, expecting one TimerSpec per timer accessed by the ParDo. I
>>> >>> suggest that we instead make TimerMap the default for
>>> portability,
>>> >>> and model the current behavior on top of timer map. If this
>>> proves
>>> >>> problematic for some runners, we could instead introduce a new
>>> >>> TimerSpec proto to represent TimerMap.
>>> >>>
>>> >>> Thoughts?
>>> >>>
>>> >>> Reuven
>>> >>>
>>>
>>
>
> --
>
> This email may be confidential and privileged. If you received this
> communication by mistake, please don't forward it to anyone else, please
> erase all copies and attachments, and please let me know that it has gone
> to the wrong person.
>
> The above terms reflect a potential business arrangement, are provided
> solely as a basis for further discussion, and are not intended to be and do
> not constitute a legally binding obligation. No legally binding obligations
> will be created, implied, or inferred until an agreement in final form is
> executed in writing by all parties involved.
>
