Thanks Dawid, having a look at CepOperator was useful. I implemented something with one difference I feel might be important:
I noticed that in the CepOperator the timer is being registered for currentWatermark+1, instead of using the event's timestamp. Is there a reason for this? I think this implies a quadratic number of triggers, on the number of keys with events that arrived after the current watermark. For instance, if you have 1000 events per second on different keys (and different timestamps), a watermark that is delayed 1 second will fire ~1 million times. Is this a requirement to the NFA implementation? Would this not be a problem? Thanks, once again. Dawid Wysakowicz <dwysakow...@apache.org> escreveu no dia segunda, 10/05/2021 à(s) 09:13: > Hey Miguel, > > I think you could take a look at the CepOperator which does pretty much > what you are describing. > > As for more direct answers for your questions. If you use > KeyedProcessFunction it is always scoped to a single Key. There is no way > to process events from other keys. If you want to have more control over > state and e.g. use PriorityQueue which would be snapshotted on checkpoint > you could look into using Operator API. Bare in mind it is a semi-public > API. It is very low level and subject to change rather frequently. Another > thing to consider is that if you use PriorityQueue instead of e.g. MapState > for buffering and ordering events you are constrained by the available > memory. We used PriorityQueue in the past in the CepOperator but migrated > it to MapState. > > It is possible that events in downstream operators can become late. It all > depends on the timestamp of the events you emit from the "sorting" > operator. If you emit records with timestamps larger than the Watermark > that "triggered" its generation it can become late. > > Hope those tips could help you a bit. > > Best, > > Dawid > On 04/05/2021 14:49, Miguel Araújo wrote: > > Hi Timo, > > Thanks for your answer. I think I wasn't clear enough in my initial > message, so let me give more details. > > The stream is not keyed by timestamp, it's keyed by a custom field (e.g., > user-id) and then fed into a KeyedProcessFunction. I want to process all > events for a given user in order, before sending them downstream for > further processing in other operators. I don't want to hold events longer > than needed, hence using the watermark to signal which events can be > processed. > I don't think your suggestion of using a ListState would work, because we > would effectively have one list per user. That would imply (among other > things) that an event could only be processed when a new event for the same > user arrives, which would not only imply a (potentially) huge latency, but > also data leakage. Not to mention that the events being sent could easily > be considered late-events to the downstream operators. > The idea of keying by timestamp was an "evolution" of the ListState > suggestion, where events-to-be-later-processed would be kept sorted in the > map (which is what would be keyed by timestamp). We could iterate the map > to process the events, instead of fetching the full list and sorting it to > process the events in order. I don't think this solves any of the problems > mentioned above, so I think that mentioning it only raised confusion. > > Regarding global event-time order, that's not really what I'm after. I > only need event-time order per key, but I want to process the event as soon > as possible, constrained by knowing that it is "safe" to do so because no > event with a smaller timestamp for this key is yet to come. > > So, rephrasing my question as I'm not sure that part was clear in the > initial message, here is the idea: > - keeping one priority queue (ordered by timestamp) in each > KeyedProcessFunction instance. Therefore, each priority queue would store > events for multiple keys. > - when an event arrives, we push it to the queue and then process events > (updating state and sending them downstream) while their timestamp is lower > than the current watermark. > > The question is: > - is this fault tolerant? The priority queue is not state that is managed > by flink, but it should be recoverable on replay. > - is it possible that the events I'm sending downstream become late-events > for a different operator, for some reason? Will they always be sent before > the watermark of the event that originated the processElement() call? > - I would effectively be processing multiple elements (from multiple keys) > in the same call to processElement(). Is there a way to access the state of > different keys? > > This doesn't feel like the right approach. Is there an operator more > suitable than a KeyedProcessFunction which would allow me to handle the > state for multiple keys in this task manager? Should I register a timer to > trigger on the event timestamp instead? I believe timers trigger on > watermarks, so that could theoretically work, although it feels a little > weird. After all, what I want is just to buffer events so that they are > only processed when the watermark has caught up to them. > > Thanks > > Timo Walther <twal...@apache.org> escreveu no dia sexta, 30/04/2021 à(s) > 12:05: > >> Hi Miguel, >> >> your initial idea sounds not too bad but why do you want to key by >> timestamp? Usually, you can simply key your stream by a custom key and >> store the events in a ListState until a watermark comes in. >> >> But if you really want to have some kind of global event-time order, you >> have two choices: >> >> - either a single operator with parallelism 1 that performs the ordering >> - or you send the every event to every operator using the broadcast >> state pattern [1] >> >> It is guaranteed that watermark will reach the downstream operator or >> sink after all events. Watermarks are synchronized across all parallel >> operator instances. You can store a Map uncheckpointed by this means >> that you have to ensure to initialize the map again during recovery. >> >> Regards, >> Timo >> >> >> [1] >> >> https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/stream/state/broadcast_state.html >> >> On 30.04.21 11:37, Miguel Araújo wrote: >> > Hi everyone, >> > >> > I have a KeyedProcessFunction whose events I would like to process in >> > event-time order. >> > My initial idea was to use a Map keyed by timestamp and, when a new >> > event arrives, iterate over the Map to process events older than the >> > current watermark. >> > >> > The issue is that I obviously can't use a MapState, as my stream is >> > keyed, so the map would be scoped to the current key. >> > Is using a "regular" (i.e., not checkpointed) Map an option, given that >> > its content will be recreated by the replay of the events on a restart? >> > Is it guaranteed that the watermark that triggered the processing of >> > multiple events (and their subsequent push downstream) is not sent >> > downstream before these events themselves? >> > >> > Thanks, >> > Miguel >> >>
