Hi Reuven,
yes, if this change is intended to be used by end users, then
DoFnSignatures cannot be used, agree on that. Regarding the relationship
with dynamic state - I agree that this is separate problem, but because
it is close enough, we should know how we want to deal with that.
Because state and timers share some functionality (after all timers need
state to be fault tolerant), these API should IMO share the same logic.
Whatever solution chosen to expose dynamic timers, it should extend to
dynamic state.
I'd like to stop a little with the premise that users want dynamic
timers (that is timers whose *name* - and therefore behavior - is
determined by incoming data). Could this case be modeled so that the
timer actually has one more (implicit) state variable that actually
holds collection of tuples (timestamp, name)? Then the timer would be
invoked at given (minimum of all currently set) timestamps with
respective name? The question here probably is - can this have
performance impact? That is to say - can any runner actually do anything
different from this in the sense of time complexity of the algorithm?
I'm a little afraid if actually letting users define data-driven timers
might not be too restrictive for some runners. Yes, runners that don't
have this option would probably be able to resort to the logic described
above, but if this work could be reasonably done for all runners, then
we wouldn't force runners to actually implement it. And, the API could
look as if the timers were actually dynamic.
Jan
P.S. If dynamic (and therefore any number) of timers can be actually
implemented using single timer, that might be interesting pattern,
because single timer per (window, key) has many nice properties, like it
implicitly avoids situation where timer invocation is not ordered
([BEAM-7520]), which seems to issue for multiple runners (samza,
portable flink).
On 10/22/19 6:52 PM, Reuven Lax 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
<mailto: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>
>>> <mailto: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
>>>
