Hi Sujit, actually, according to https://ci.apache.org/projects/flink/flink-docs-release-1.2/dev/ windows.html#allowed-lateness the sliding window should fire each time for each element arriving late.
Did you set the following for your window operator?
.window(<window assigner>)
.allowedLateness(<time>)
The expected behaviour should be:
1) fire once a watermark exceeding the sliding window end has been received
2) fire each time an event after the sliding window end is received as long as
no watermark is exceeding sliding window end + allowed lateness
In your case, if I see it correctly and do not misinterpret your example, the
late firing has nothing to do with the sliding window but rather your
BoundedOutOfOrdernessGenerator:
Since a sliding window is only closed and fired once the watermark is received,
its first time firing is actually 5s behind as per your maxOutOfOrderness.
It actually looks like you may be better of using allowed lateness for your
usecase depending on what you actually need.
Regards
Nico
On Monday, 27 February 2017 15:44:20 CET Sujit Sakre wrote:
> Hi,
>
> Hope you are well.
>
> We have encountered an issue in processing sliding windows. Here we have
> encountered the problem that if the last record is outside of the sliding
> window end time then it does not process the record till the next sliding
> window is completely occupied and gets triggered.
>
> Please consider the example below:
>
> The datastream is Kafka based and Keyed on Location (1, 2, 3, 4 etc.)
>
> Our previous sliding window
> starts at
> 09-09-2016 12:54:00,
> End at
> 09-09-2016 13:00:00
> *Key is 15 (Location)*
>
> Records in between with timestamps:
> 09-09-2016 12:56:33
> 09-09-2016 12:56:47
> 09-09-2016 12:58:04
> 09-09-2016 12:58:39
> 09-09-2016 12:58:45
>
> However the next window starts at
> 09-09-2016 13:04:00
> and ends at
> 09-09-2016 13:10:00
> *with Key as 16 (Location)*
> with record timestamps:
>
> 09-09-2016 13:04:48
> 09-09-2016 13:06:07
> 09-09-2016 13:06:38
> 09-09-2016 13:07:25
> 09-09-2016 13:08:00
> 09-09-2016 13:08:20
> 09-09-2016 13:08:38
>
>
> is not processed until records are entered in *Location 17* with the
> timestamps:
> 09-09-2016 13:08:48
> 09-09-2016 13:08:55
> 09-09-2016 13:09:11
> 09-09-2016 13:11:48
> The window that gets formed at that time has
> Start Time: 09-09-2016 13:06:00
> End Time: 09-09-2016 13:12:00
>
> We are using the standard BoundedOutOfOrdernessGenerator with maximum out
> of orderness of 5 seconds (we have tried various other combinations of the
> maxoutoforderness values but without success), and Event Time based
> processing.
>
> /** * This generator generates watermarks assuming that elements
> arrive out of order, * but only to a certain degree. The latest
> elements for a certain timestamp t will arrive * at most n
> milliseconds after the earliest elements for timestamp t. */public
> class BoundedOutOfOrdernessGenerator extends
> AssignerWithPeriodicWatermarks<MyEvent> {
>
> private final long maxOutOfOrderness = 5000; // 5 seconds
>
> private long currentMaxTimestamp;
>
> @Override
> public long extractTimestamp(MyEvent element, long
> previousElementTimestamp) {
> long timestamp = element.getCreationTime();
> currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
> return timestamp;
> }
>
> @Override
> public Watermark getCurrentWatermark() {
> // return the watermark as current highest timestamp minus the
> out-of-orderness bound
> return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
> }}
>
>
> Thus to summarize the problem,
>
> the records in the window before the last window with different location
> are not processed until there is a window next with a timestamp that is
> more than the End Time timestamp of the existing sliding window.
> This means the window is not processed until next set of records arrive
> with timestamps that are more than existing end time window timestamp.
>
> In a real situation, this means that we wait till the next set of records
> arrive, which may be after a very long duration (e.g. maybe 1 hour), and
> previous records are not processed till then.
>
> Is this a problem that is by behavior?
> Why does the sliding window not process the record that is present even
> after not receiving a record for more than a substantial amount of time,
> e.g. 30 minutes?
> How do we resolve this situation?
>
> Please could you suggest how to resolve this.
>
> Many thanks.
>
>
>
> *Sujit Sakre*
signature.asc
Description: This is a digitally signed message part.
