flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Aljoscha Krettek <aljos...@apache.org>
Subject Re: Event-Time Windowing
Date Wed, 07 Oct 2015 15:51:11 GMT
Hi,
yes, once this PR is merged https://github.com/apache/flink/pull/1238 you
can switch between time characteristics and also use the aggregations
functions such as sum(...). I'm hoping to merge this by tonight. The tests
are still running right now. :D

Cheers,
Aljoscha



On Wed, 7 Oct 2015 at 17:45 Alexander Kolb <alexander.kolb@mni.fh-giessen.de>
wrote:

> Thanks!
>
> This works with the exception that I have to use the reduceWindow() method
> when summing up my the content of the window.
> There still seems to be some work to do.
>
> With the finished Api will I be able to switch from event-time to
> processing- or ingestion-time without having to adjust my code?
>
> Best,
> Alex
>
> Aljoscha Krettek <aljoscha@apache.org> schrieb am Mi., 7. Okt. 2015,
> 17:23:
>
>> Hi,
>> right now, the 0.10-SNAPSHOT is in a bit of a weird state. We still have
>> the old windowing API in there alongside the new one. To make your example
>> use the new API that actually uses the timestamps and watermarks you would
>> use the following code:
>>
>> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>> val stream = env.addSource(SourceWithEventTime)
>>
>> stream
>>   .timeWindowAll(Time.of(5,TimeUnit.SECONDS))
>>   // or .windowAll(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS)))
>>   .sum(0)
>>   .print()
>>
>> the version for keyed streams would be:
>> stream
>>   .keyBy(...)
>>   .timeWindow(Time.of(5,TimeUnit.SECONDS))
>>   // or .window(SlidingTimeWindows.of(Time.of(5, TimeUnit.SECONDS)))
>>   .sum(0)
>>   .print()
>>
>> I hope this helps. :D
>>
>> Cheers,
>> Aljoscha
>>
>>
>> On Wed, 7 Oct 2015 at 16:54 Alexander Kolb <
>> alexander.kolb@mni.fh-giessen.de> wrote:
>>
>>> Hi Guys,
>>>
>>> I'm trying to use the event-time windowing feature. But the windowing
>>> does not work as expected.
>>>
>>> What I've been doing is to write my own source which implements the
>>> EventTimeSourceFunction and uses the collectWithTimeStamp  method.
>>> Additionally I'm emitting a watermark after each element.
>>>
>>> My job to test this looks like this:
>>>
>>> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
>>> val stream = env.addSource(SourceWithEventTime)
>>>
>>> stream
>>>   .window(Time.of(5,TimeUnit.SECONDS))
>>>   .sum(0)
>>>   .flatten()
>>>   .print()
>>>
>>> env.execute()
>>>
>>> The Input are some tuples with TimeStamps set 10 seconds apart:
>>>
>>> value: (1,test) timestamp: 1444228980390
>>> value: (2,foo) timestamp: 1444228990390
>>> value: (3,bar) timestamp: 1444229000390
>>>
>>> What I'm expecting is that each tuple goes into a separate window.
>>> The actual output is the sum of all tuples, hence all tuples are
>>> collected in the same window.
>>>
>>> Thanks in advance!
>>> Alex
>>>
>>>
>>>

Mime
View raw message