flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Jason Brelloch <jb.bc....@gmail.com>
Subject Re: Generate timestamps in front of event for event time windows
Date Thu, 04 Aug 2016 13:05:37 GMT
Thanks Aljoscha,

Looking forward to the 1.1. release.  I managed to solve my problem using
this example code:

https://bitbucket.org/snippets/vstoyak/o9Rqp
(courtesy of Vladimir Stoyak)

I had to create a custom window and window assigner.  Hopefully that will
help someone else.

On Wed, Aug 3, 2016 at 8:35 PM, Aljoscha Krettek <aljoscha@apache.org>
wrote:

> Hi,
> a watermark cannot be sent before the element that makes you send that
> watermark. A watermark of time T tells the system that no element will
> arrive in the future with timestamp T or less, thus you cannot send it
> before. It seems that what you are trying to achieve can be solved by using
> session windows, which will be part of the upcoming 1.1 release:
> https://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/windows.html#session-windows
>
> Cheer,
> Aljoscha
>
> On Wed, 3 Aug 2016 at 12:19 Jason Brelloch <jb.bc.flk@gmail.com> wrote:
>
>> A little more info.  Here is a simplified version of my
>> trigger: (windowConfiguration.timespan is the duration of the window)
>>
>> class CustomTrigger extends Trigger[QualifiedEvent, Window] {
>>
>>   val stateTimeDescr = new
>> ValueStateDescriptor[Long]("relevantTimestamp", classOf[Long], 0)
>>
>>   override def onElement(event: QualifiedEvent, timestamp: Long, W:
>> Window, ctx: TriggerContext): TriggerResult = {
>>
>>     val relevantTimestamp = ctx.getPartitionedState(stateTimeDescr)
>>     val windowConfigurationState =
>> ctx.getPartitionedState(windowConfigDescr)
>>     var windowConfiguration = windowConfigurationState.value()
>>     if(windowConfiguration == null) {
>>       windowConfigurationState.update(event.alertConfiguration.window.get)
>>       windowConfiguration = event.alertConfiguration.window.get
>>     }
>>
>>     if(relevantTimestamp.value() == 0) {
>>       ctx.registerEventTimeTimer(event.event.created.toEpochMilli +
>> windowConfiguration.timespan.toMillis)
>>       relevantTimestamp.update (event.event.created.toEpochMilli +
>> windowConfiguration.timespan.toMillis)
>>     }
>>
>>     TriggerResult.CONTINUE
>>   }
>>
>>   override def onEventTime(timestamp: Long, W: Window, ctx:
>> TriggerContext): TriggerResult = {
>>     TriggerResult.FIRE_AND_PURGE
>>   }
>>
>>   override def onProcessingTime(timestamp: Long, W: Window, ctx:
>> TriggerContext): TriggerResult = {
>>     TriggerResult.CONTINUE
>>   }
>> }
>>
>> And here is the actual window execution:
>>
>> val stream = env.fromCollection(inputEvents)
>>         .assignAscendingTimestamps((e: QualifiedEvent) => {
>> e.event.created.toEpochMilli })
>>         .keyBy((e: QualifiedEvent) => {
>> e.alertConfiguration.alertId.toString })
>>         .window(GlobalWindows.create)
>>         .trigger(ConfigurableTrigger.create)
>>         .apply(new GrouperFunction).name("Grouper Function")
>>
>> Oddly enough when I do this with just a basic window function it works
>> and I only get the two events I am supposed to:
>>
>> val stream = env.fromCollection(inputEvents)
>>         .assignAscendingTimestamps((e: QualifiedEvent) => {
>> e.event.created.toEpochMilli })
>>         .keyBy((e: QualifiedEvent) => {
>> e.alertConfiguration.alertId.toString })
>>         .timeWindow(Time.minutes(5))
>>         .apply(new GrouperFunction).name("Grouper Function")
>>
>>
>> On Wed, Aug 3, 2016 at 2:29 PM, Jason Brelloch <jb.bc.flk@gmail.com>
>> wrote:
>>
>>> Hey guys,
>>>
>>> I am trying to use event time along with a custom window to capture a
>>> subset of events.  The problem I am running into is that it seems that
>>> event that generates the timestamp/watermark arrives in the window before
>>> the onEventTime() call is made that closes the window.  Example:
>>>
>>> Window is supposed to capture 5 minutes of events after first event
>>> arrives
>>> Event 1: timestamp 12:01 - registers event timer for 12:06
>>> Event 2: timestamp 12:03
>>> Event 3: timestamp 12:20 - fires and purges window
>>>
>>> I get all three events in the window, instead of just the two the are
>>> really within the 5 minute window.
>>>
>>> Is there someway to force the timestamp to arrive in the window before
>>> the event that generated it?
>>>
>>> Thanks!
>>>
>>> --
>>> *Jason Brelloch* | Product Developer
>>> 3405 Piedmont Rd. NE, Suite 325, Atlanta, GA 30305
>>> <http://www.bettercloud.com/>
>>> Subscribe to the BetterCloud Monitor
>>> <https://www.bettercloud.com/monitor?utm_source=bettercloud_email&utm_medium=email_signature&utm_campaign=monitor_launch>
-
>>> Get IT delivered to your inbox
>>>
>>
>>
>>
>> --
>> *Jason Brelloch* | Product Developer
>> 3405 Piedmont Rd. NE, Suite 325, Atlanta, GA 30305
>> <http://www.bettercloud.com/>
>> Subscribe to the BetterCloud Monitor
>> <https://www.bettercloud.com/monitor?utm_source=bettercloud_email&utm_medium=email_signature&utm_campaign=monitor_launch>
-
>> Get IT delivered to your inbox
>>
>


-- 
*Jason Brelloch* | Product Developer
3405 Piedmont Rd. NE, Suite 325, Atlanta, GA 30305
<http://www.bettercloud.com/>
Subscribe to the BetterCloud Monitor
<https://www.bettercloud.com/monitor?utm_source=bettercloud_email&utm_medium=email_signature&utm_campaign=monitor_launch>
-
Get IT delivered to your inbox

Mime
View raw message