You are viewing a plain text version of this content. The canonical link for it is here.

Posted to user@flink.apache.org by Paul Joireman <pa...@physiq.com> on 2016/08/30 19:28:29 UTC

Setting EventTime window width using stream data

Hi all,


Is it possible to dynamically set the size/width of a SlidingEventTimeWindow based on a data coming from the stream?   Our use case is as follows.   We create a

stream sourced from external system and coming in as a JSON string which is deserialized to a stream of POJO.    The deserialized object contains an event timestamp,

data and details about how to analyze the contained data as well as the length of a time window to analyze.    It would be ideal if we could leverage the functionality of the

SlidingEventTimeWindows but instead of hard coding the window times, use data from the message to configure this on the fly.   Is this possible?


Note: the stream is keyed and timestamped with event time and the window size will not change for the same key.


Ultimately we need a way to dynamically change the window sizes in order to adjust to different timing specifications not directly controlled or even known before-hand by

the person writing the flink analysis program.


Paul

Re: Setting EventTime window width using stream data

Posted by Paul Joireman <pa...@physiq.com>.

Thanks Aljoscha,


The elements can certainly be configured to have the window size and slide as part of them and that will not change for the same elements.   There may a different class of elements with a different window or slide but those values will be essentially final for that class of elements.


Paul

________________________________
From: Aljoscha Krettek <al...@apache.org>
Sent: Wednesday, August 31, 2016 10:53:17 AM
To: user@flink.apache.org
Cc: Chad Conkright
Subject: Re: Setting EventTime window width using stream data

Just checking, all the elements that would fall into a window of length X also have X as a property? In that case you should be able to do something like this:

public Collection<TimeWindow> assignWindows(PojoType element, long timestamp, WindowAssignerContext context) {
    long size = element.windowSize;
    long slide = element.windowSlide;

    if (timestamp > Long.MIN_VALUE) {
        List<TimeWindow> windows = new ArrayList<>((int) (size / slide));
        long lastStart = TimeWindow.getWindowStartWithOffset(timestamp, offset, slide);
        for (long start = lastStart;
            start > timestamp - size;
            start -= slide) {
            windows.add(new TimeWindow(start, start + size));
        }
        return windows;
    } else {
        throw new RuntimeException("Record has Long.MIN_VALUE timestamp (= no timestamp marker). " +
                "Is the time characteristic set to 'ProcessingTime', or did you forget to call " +
                "'DataStream.assignTimestampsAndWatermarks(...)'?");
    }
}

this is basically a copy of SlidingEventTimeWindows where assignWindows is changed for the above and where size/slide are not stored in the assigner but are read from the object. This only works if all elements of a key that should fall into the same windows have the same size/slide property, otherwise they would spawn different windows.

Cheers,
Aljoscha

On Tue, 30 Aug 2016 at 21:28 Paul Joireman <pa...@physiq.com>> wrote:

Hi all,


Is it possible to dynamically set the size/width of a SlidingEventTimeWindow based on a data coming from the stream?   Our use case is as follows.   We create a

stream sourced from external system and coming in as a JSON string which is deserialized to a stream of POJO.    The deserialized object contains an event timestamp,

data and details about how to analyze the contained data as well as the length of a time window to analyze.    It would be ideal if we could leverage the functionality of the

SlidingEventTimeWindows but instead of hard coding the window times, use data from the message to configure this on the fly.   Is this possible?


Note: the stream is keyed and timestamped with event time and the window size will not change for the same key.


Ultimately we need a way to dynamically change the window sizes in order to adjust to different timing specifications not directly controlled or even known before-hand by

the person writing the flink analysis program.


Paul

Re: Setting EventTime window width using stream data

Posted by Aljoscha Krettek <al...@apache.org>.

Just checking, all the elements that would fall into a window of length X
also have X as a property? In that case you should be able to do something
like this:

public Collection<TimeWindow> assignWindows(PojoType element, long
timestamp, WindowAssignerContext context) {
    long size = element.windowSize;
    long slide = element.windowSlide;

    if (timestamp > Long.MIN_VALUE) {
        List<TimeWindow> windows = new ArrayList<>((int) (size / slide));
        long lastStart = TimeWindow.getWindowStartWithOffset(timestamp,
offset, slide);
        for (long start = lastStart;
            start > timestamp - size;
            start -= slide) {
            windows.add(new TimeWindow(start, start + size));
        }
        return windows;
    } else {
        throw new RuntimeException("Record has Long.MIN_VALUE timestamp (=
no timestamp marker). " +
                "Is the time characteristic set to 'ProcessingTime', or did
you forget to call " +
                "'DataStream.assignTimestampsAndWatermarks(...)'?");
    }
}

this is basically a copy of SlidingEventTimeWindows where assignWindows is
changed for the above and where size/slide are not stored in the assigner
but are read from the object. This only works if all elements of a key that
should fall into the same windows have the same size/slide property,
otherwise they would spawn different windows.

Cheers,
Aljoscha

On Tue, 30 Aug 2016 at 21:28 Paul Joireman <pa...@physiq.com> wrote:

> Hi all,
>
>
> Is it possible to dynamically set the size/width of a
> SlidingEventTimeWindow based on a data coming from the stream?   Our use
> case is as follows.   We create a
>
> stream sourced from external system and coming in as a JSON string which
> is deserialized to a stream of POJO.    The deserialized object contains an
> event timestamp,
>
> data and details about how to analyze the contained data as well as the
> length of a time window to analyze.    It would be ideal if we could
> leverage the functionality of the
>
> SlidingEventTimeWindows but instead of hard coding the window times, use
> data from the message to configure this on the fly.   Is this possible?
>
>
> Note: the stream is keyed and timestamped with event time and the window
> size will not change for the same key.
>
>
> Ultimately we need a way to dynamically change the window sizes in order
> to adjust to different timing specifications not directly controlled or
> even known before-hand by
>
> the person writing the flink analysis program.
>
>
> Paul
>
>
>