Hi,
This is mostly correct, but you cannot register a timer in open() because we
don't have an active key there. Only in process() and onTimer() can you
register a timer.
In your case, I would suggest to somehow clamp the timestamp to the nearest 2
minute (or whatever) interval or to keep an extra ValueState that tells you
whether you already registered a timer.
Best,
Aljoscha
> On 5. Sep 2017, at 16:55, Kien Truong <duckientru...@gmail.com> wrote:
>
> Hi,
>
> You can register a processing time timer inside the onTimer and the open
> function to have a timer that run periodically.
> Pseudo-code example:
>
> ValueState<Long> lastRuntime;
>
> void open() {
> ctx.timerService().registerProcessingTimeTimer(current.timestamp + 60000);
> }
>
> void onTimer() {
> // Run the periodic task
> if (lastRuntime.get() + 60000 == timeStamp) {
> periodicTask();
> }
> // Re-register the processing time timer timer
> lastRuntime.setValue(timeStamp);
> ctx.timerService().registerProcessingTimeTimer(current.timestamp + 60000);
> }
>
> void periodicTask()
>
> For the second question, timer are already scoped by key, so you can keep a
> lastModified variable as a ValueState,
> then compare it to the timestamp provided by the timer to see if the current
> key should be evicted.
> Checkout the example on the ProcessFunction page.
>
> https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/stream/process_function.html
>
> <https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/stream/process_function.html>
>
> Best regards,
> Kien
>
> On 9/5/2017 11:49 AM, Navneeth Krishnan wrote:
>> Hi All,
>>
>> I have a streaming pipeline which is keyed by userid and then to a flatmap
>> function. I need to clear the state after sometime and I was looking at
>> process function for it.
>>
>> Inside the process element function if I register a timer wouldn't it create
>> a timer for each incoming message?
>> // schedule the next timer 60 seconds from the current event time
>> ctx.timerService().registerEventTimeTimer(current.timestamp + 60000);
>> How can I get something like a clean up task that runs every 2 mins and
>> evicts all stale data? Also is there a way to get the key inside onTimer
>> function so that I know which key has to be evicted?
>>
>> Thanks,
>> Navneeth
