Hi, Probably another solution would be to register a timer (using KeyedProcessFunction) once we see an element after keyBy. The timer will fire in windowIntervalMs. Upon firing, it will emit a dummy element which will be ignored (or subtracted) in the end. Upon receiving each new element, the function will shift the timer accordingly.
Regards, Roman On Mon, Feb 8, 2021 at 10:50 AM Jan Brusch <jan.bru...@neuland-bfi.de> wrote: > Hi Yun, > > thanks for your reply. > > I do agree with your point about standard windows being for high level > operations and the lower-level apis offering a rich toolset for most > advanced use cases. > > I have tried to solve my problem with keyedProcessFunctions also but was > not able to get it to work for two reasons: > > 1) I was not able to set up a combination of ValueState, Timers and > Triggers that emulated a sliding window with a rising and falling count > (including 0) good enough. > > 2) Memory Leak: States / Windows should be cleared after a certain time of > being at count 0 in order to prevent an infinitely rising of ValueStates > (that are not needed anymore) > > > Can you maybe please elaborate in pseudocode how you would envision your > solution? > > > Best regards > > Jan > On 08.02.21 05:31, Yun Gao wrote: > > Hi Jan, > > From my view, I think in Flink Window should be as a "high-level" > operation for some kind > of aggregation operation and if it could not satisfy the requirements, we > could at least turn to > using the "low-level" api by using KeyedProcessFunction[1]. > > In this case, we could use a ValueState to store the current value for > each key, and increment > the value on each element. Then we could also register time for each key > on receiving the first > element for this key, and in the onTimer callback, we could send the > current state value, update > the value to 0 and register another timer for this key after 30s. > > Best, > Yun > > > > [1] > https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/process_function.html#the-keyedprocessfunction > > ------------------Original Mail ------------------ > *Sender:*Jan Brusch <jan.bru...@neuland-bfi.de> > <jan.bru...@neuland-bfi.de> > *Send Date:*Sat Feb 6 23:44:00 2021 > *Recipients:*user <user@flink.apache.org> <user@flink.apache.org> > *Subject:*Sliding Window Count: Tricky Edge Case / Count Zero Problem > >> Hi, >> I was recently working on a problem where we wanted to implement a >> simple count on a sliding window, e.g. "how many messages of a certain >> type were emitted by a certain type of sensor in the last n minutes". >> Which sounds simple enough in theory: >> >> messageStream >> .keyBy(//EmitterType + MessageType) >> .assignWindow(SlidingProcessingTimeWindows.of(Time.minutes(n), >> Time.seconds(30))) >> .map(_ => 1) >> .reduce((x,y) => x + y) >> .addSink(...) >> >> But there is a tricky edge case: The downstream systems will never know >> when the count for a certain key goes back to 0, which is important for >> our use case. The technical reason being that flink doesn't open a >> window if there are no entries, i.e. a window with count 0 doesn't exist >> in flink. >> >> We came up with the following solution for the time being: >> >> messageStream >> .keyBy(//EmitterType + MessageType) >> .window(GlobalWindows.create()) >> .trigger(ContinuousEventTimeTrigger.of(Time.seconds(30))) >> .evictor(// CustomEvictor: Evict all messages older than n minutes >> BEFORE processing the window) >> .process(// CustomCounter: Count all Messages in Window State); >> .addSink(...) >> >> In the case of zero messages in the last n minutes, all messages will be >> evicted from the window and the process-function will get triggered one >> last time on the now empty window, so we can produce a count of 0. >> >> I have two problems, though, with this solution: >> 1) It is computationally inefficient for a simple count, as custom >> process functions will always keep all messages in state. And, on every >> trigger all elements will have to be touched twice: To compare the >> timestamp and to count. >> 2) It does seem like a very roundabout solution to a simple problem. >> >> So, I was wondering if there was a more efficient or "flink-like" >> approach to this. Sorry for the long writeup, but I would love to hear >> your takes. >> >> >> Best regards >> Jan >> >> -- >> neuland – Büro für Informatik GmbH >> Konsul-Smidt-Str. 8g, 28217 Bremen >> >> Telefon (0421) 380107 57 >> Fax (0421) 380107 99 >> https://www.neuland-bfi.de >> >> https://twitter.com/neuland >> https://facebook.com/neulandbfi >> https://xing.com/company/neulandbfi >> >> >> Geschäftsführer: Thomas Gebauer, Jan Zander >> Registergericht: Amtsgericht Bremen, HRB 23395 HB >> USt-ID. DE 246585501 > > -- > neuland – Büro für Informatik GmbH > Konsul-Smidt-Str. 8g, 28217 Bremen > > Telefon (0421) 380107 57 > Fax (0421) 380107 99https://www.neuland-bfi.de > https://twitter.com/neulandhttps://facebook.com/neulandbfihttps://xing.com/company/neulandbfi > > > Geschäftsführer: Thomas Gebauer, Jan Zander > Registergericht: Amtsgericht Bremen, HRB 23395 HB > USt-ID. DE 246585501 > >
