| Transformation | -Description | -
|---|---|
| Tumbling time window KeyedStream → WindowedStream |
-
- - Defines a window of 5 seconds, that "tumbles". This means that elements are - grouped according to their timestamp in groups of 5 second duration, and every element belongs to exactly one window. - The notion of time is specified by the selected TimeCharacteristic (see time). - {% highlight java %} -keyedStream.timeWindow(Time.seconds(5)); - {% endhighlight %} - - |
-
| Sliding time window KeyedStream → WindowedStream |
-
- - Defines a window of 5 seconds, that "slides" by 1 second. This means that elements are - grouped according to their timestamp in groups of 5 second duration, and elements can belong to more than - one window (since windows overlap by at most 4 seconds) - The notion of time is specified by the selected TimeCharacteristic (see time). - {% highlight java %} -keyedStream.timeWindow(Time.seconds(5), Time.seconds(1)); - {% endhighlight %} - - |
-
| Tumbling count window KeyedStream → WindowedStream |
-
- - Defines a window of 1000 elements, that "tumbles". This means that elements are - grouped according to their arrival time (equivalent to processing time) in groups of 1000 elements, - and every element belongs to exactly one window. - {% highlight java %} -keyedStream.countWindow(1000); - {% endhighlight %} - - |
-
| Sliding count window KeyedStream → WindowedStream |
-
- - Defines a window of 1000 elements, that "slides" every 100 elements. This means that elements are - grouped according to their arrival time (equivalent to processing time) in groups of 1000 elements, - and every element can belong to more than one window (as windows overlap by at most 900 elements). - {% highlight java %} -keyedStream.countWindow(1000, 100) - {% endhighlight %} - - |
-
| Transformation | -Description | -
|---|---|
| Tumbling time window KeyedStream → WindowedStream |
-
- - Defines a window of 5 seconds, that "tumbles". This means that elements are - grouped according to their timestamp in groups of 5 second duration, and every element belongs to exactly one window. - The notion of time is specified by the selected TimeCharacteristic (see time). - {% highlight scala %} -keyedStream.timeWindow(Time.seconds(5)) - {% endhighlight %} - - |
-
| Sliding time window KeyedStream → WindowedStream |
-
- - Defines a window of 5 seconds, that "slides" by 1 second. This means that elements are - grouped according to their timestamp in groups of 5 second duration, and elements can belong to more than - one window (since windows overlap by at most 4 seconds) - The notion of time is specified by the selected TimeCharacteristic (see time). - {% highlight scala %} -keyedStream.timeWindow(Time.seconds(5), Time.seconds(1)) - {% endhighlight %} - - |
-
| Tumbling count window KeyedStream → WindowedStream |
-
- - Defines a window of 1000 elements, that "tumbles". This means that elements are - grouped according to their arrival time (equivalent to processing time) in groups of 1000 elements, - and every element belongs to exactly one window. - {% highlight scala %} -keyedStream.countWindow(1000) - {% endhighlight %} - - |
-
| Sliding count window KeyedStream → WindowedStream |
-
- - Defines a window of 1000 elements, that "slides" every 100 elements. This means that elements are - grouped according to their arrival time (equivalent to processing time) in groups of 1000 elements, - and every element can belong to more than one window (as windows overlap by at most 900 elements). - {% highlight scala %} -keyedStream.countWindow(1000, 100) - {% endhighlight %} - - |
-
+
+### Tumbling Windows
+
+A *tumbling windows* assigner assigns elements to fixed length, non-overlapping windows of a
+specified *window size*.. For example, if you specify a window size of 5 minutes, the window
+function will get 5 minutes worth of elements in each invocation.
-The general mechanism can define more powerful windows at the cost of more verbose syntax. For example,
-below is a window definition where windows hold elements of the last 5 seconds and slides every 1 second,
-but the execution of the window function is triggered when 100 elements have been added to the
-window, and every time execution is triggered, 10 elements are retained in the window:
+
+
+### Sliding Windows
+
+The *sliding windows* assigner assigns elements to windows of fixed length equal to *window size*,
+as the tumbling windows assigner, but in this case, windows can be overlapping. The size of the
+overlap is defined by the user-specified parameter *window slide*. As windows are overlapping, an
+element can be assigned to multiple windows
+
+For example, you could have windows of size 10 minutes that slide by 5 minutes. With this you get 10
+minutes worth of elements in each invocation of the window function and it will be invoked for every
+5 minutes of data.
+
+
+
+### Session Windows
+
+The *session windows* assigner is ideal for cases where the window boundaries need to adjust to the
+incoming data. Both the *tumbling windows* and *sliding windows* assigner assign elements to windows
+that start at fixed time points and have a fixed *window size*. With session windows it is possible
+to have windows that start at individual points in time for each key and that end once there has
+been a certain period of inactivity. The configuration parameter is the *session gap* that specifies
+how long to wait for new data before considering a session as closed.
+
+
+
+### Specifying a Window Assigner
+
+The built-in window assigners (except `GlobalWindows`) come in two versions. One for processing-time
+windowing and one for event-time windowing. The processing-time assigners assign elements to
+windows based on the current clock of the worker machines while the event-time assigners assign
+windows based on the timestamps of elements. Please have a look at
+[event time](/apis/streaming/event_time.html) to learn about the difference between processing time
+and event time and about how timestamps can be assigned to elements.
+
+The following code snippets show how each of the window assigners can be used in a program:
| Transformation | -Description | -
|---|---|
| Global window KeyedStream → WindowedStream |
-
- - All incoming elements of a given key are assigned to the same window. - The window does not contain a default trigger, hence it will never be triggered - if a trigger is not explicitly specified. - - {% highlight java %} -stream.window(GlobalWindows.create()); - {% endhighlight %} - |
-
| Tumbling event-time windows KeyedStream → WindowedStream |
-
- - Incoming elements are assigned to a window of a certain size (1 second below) based on - their timestamp. Windows do not overlap, i.e., each element is assigned to exactly one window. - This assigner comes with a default trigger that fires for a window when a - watermark with value higher than its end-value is received. - - {% highlight java %} -stream.window(TumblingEventTimeWindows.of(Time.seconds(1))); - {% endhighlight %} - |
-
| Sliding event-time windows KeyedStream → WindowedStream |
-
- - Incoming elements are assigned to a window of a certain size (5 seconds below) based on - their timestamp. Windows "slide" by the provided value (1 second in the example), and hence - overlap. This assigner comes with a default trigger that fires for a window when a - watermark with value higher than its end-value is received. - - {% highlight java %} -stream.window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1))); - {% endhighlight %} - |
-
| Tumbling processing time windows KeyedStream → WindowedStream |
-
- - Incoming elements are assigned to a window of a certain size (1 second below) based on - the current processing time. Windows do not overlap, i.e., each element is assigned to exactly one window. - This assigner comes with a default trigger that fires for a window a window when the current - processing time exceeds its end-value. - - {% highlight java %} -stream.window(TumblingProcessingTimeWindows.of(Time.seconds(1))); - {% endhighlight %} - |
-
| Sliding processing time windows KeyedStream → WindowedStream |
-
- - Incoming elements are assigned to a window of a certain size (5 seconds below) based on - their timestamp. Windows "slide" by the provided value (1 second in the example), and hence - overlap. This assigner comes with a default trigger that fires for a window a window when the current - processing time exceeds its end-value. - - {% highlight java %} -stream.window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1))); - {% endhighlight %} - |
-
| Event-time Session windows KeyedStream → WindowedStream |
-
- - Incoming elements are assigned to sessions based on a session gap interval (5 seconds in the example below). - Elements whose timestamp differs by more than the session gap are assigned to different sessions. If there are - consecutive elements which are less than the session gap apart then these will also be put into the same session, i.e. elements - can be connected into a session by intermediate elements. - - {% highlight scala %} -keyedStream.window(EventTimeSessionWindows.withGap(Time.seconds(5))); - {% endhighlight %} - |
-
| Processing time Session windows KeyedStream → WindowedStream |
-
- - This is similar to event-time session windows but works on the current processing - time instead of the timestamp of elements - - {% highlight scala %} -keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5))); - {% endhighlight %} - |
-
| Transformation | -Description | -
|---|---|
| Global window KeyedStream → WindowedStream |
-
- - All incoming elements of a given key are assigned to the same window. - The window does not contain a default trigger, hence it will never be triggered - if a trigger is not explicitly specified. - - {% highlight scala %} -stream.window(GlobalWindows.create) - {% endhighlight %} - |
-
| Tumbling event-time windows KeyedStream → WindowedStream |
-
- - Incoming elements are assigned to a window of a certain size (1 second below) based on - their timestamp. Windows do not overlap, i.e., each element is assigned to exactly one window. - This assigner comes with a default trigger that fires for a window when a - watermark with value higher than its end-value is received. - - {% highlight scala %} -stream.window(TumblingEventTimeWindows.of(Time.seconds(1))) - {% endhighlight %} - |
-
| Sliding event-time windows KeyedStream → WindowedStream |
-
- - Incoming elements are assigned to a window of a certain size (5 seconds below) based on - their timestamp. Windows "slide" by the provided value (1 second in the example), and hence - overlap. This assigner comes with a default trigger that fires for a window when a - watermark with value higher than its end-value is received. - - {% highlight scala %} -stream.window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1))) - {% endhighlight %} - |
-
| Tumbling processing time windows KeyedStream → WindowedStream |
-
- - Incoming elements are assigned to a window of a certain size (1 second below) based on - the current processing time. Windows do not overlap, i.e., each element is assigned to exactly one window. - This assigner comes with a default trigger that fires for a window a window when the current - processing time exceeds its end-value. - - - {% highlight scala %} -stream.window(TumblingProcessingTimeWindows.of(Time.seconds(1))) - {% endhighlight %} - |
-
| Sliding processing time windows KeyedStream → WindowedStream |
-
- - Incoming elements are assigned to a window of a certain size (5 seconds below) based on - their timestamp. Windows "slide" by the provided value (1 second in the example), and hence - overlap. This assigner comes with a default trigger that fires for a window a window when the current - processing time exceeds its end-value. - - {% highlight scala %} -stream.window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1))) - {% endhighlight %} - |
-
| Event-time Session windows KeyedStream → WindowedStream |
-
- - Incoming elements are assigned to sessions based on a session gap interval (5 seconds in the example below). - Elements whose timestamp differs by more than the session gap are assigned to different sessions. If there are - consecutive elements which are less than the session gap apart then these will also be put into the same session, i.e. elements - can be connected into a session by intermediate elements. - - {% highlight scala %} -keyedStream.window(EventTimeSessionWindows.withGap(Time.seconds(5))) - {% endhighlight %} - |
-
| Processing time Session windows KeyedStream → WindowedStream |
-
- - This is similar to event-time session windows but works on the current processing - time instead of the timestamp of elements - - {% highlight scala %} -keyedStream.window(ProcessingTimeSessionWindows.withGap(Time.seconds(5))) - {% endhighlight %} - |
-
| Transformation | -Description | -
|---|---|
| Processing time trigger | -
- - A window is fired when the current processing time exceeds its end-value. - The elements on the triggered window are henceforth discarded. - {% highlight java %} -windowedStream.trigger(ProcessingTimeTrigger.create()); +DataStream |
-
| Watermark trigger | -
- - A window is fired when a watermark with value that exceeds the window's end-value has been received. - The elements on the triggered window are henceforth discarded. - -{% highlight java %} -windowedStream.trigger(EventTimeTrigger.create()); -{% endhighlight %} - |
-
| Continuous processing time trigger | -
- - A window is periodically considered for being fired (every 5 seconds in the example). - The window is actually fired only when the current processing time exceeds its end-value. - The elements on the triggered window are retained. - -{% highlight java %} -windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(5))); -{% endhighlight %} - |
-
| Continuous watermark time trigger | -
- - A window is periodically considered for being fired (every 5 seconds in the example). - A window is actually fired when a watermark with value that exceeds the window's end-value has been received. - The elements on the triggered window are retained. - -{% highlight java %} -windowedStream.trigger(ContinuousEventTimeTrigger.of(Time.seconds(5))); + + +
+{% highlight scala %}
+val input: DataStream[(String, Long)] = ...
+
+input
+ .keyBy( |
-
| Count trigger | -
- - A window is fired when it has more than a certain number of elements (1000 below). - The elements of the triggered window are retained. - + + + +A `FoldFunction` specifies how elements from the input will be added to an initial +accumulator value (`""`, the empty string, in our example). This example will compute +a concatenation of all the `Long` fields of the input. + +### WindowFunction - The Generic Case + +Using a `WindowFunction` provides most flexibility, at the cost of performance. The reason for this +is that elements cannot be incrementally aggregated for a window and instead need to be buffered +internally until the window is considered ready for processing. A `WindowFunction` gets an +`Iterable` containing all the elements of the window being processed. The signature of +`WindowFunction` is this: + +
+
{% highlight java %}
-windowedStream.trigger(CountTrigger.of(1000));
+public interface WindowFunction |
-
| Purging trigger | -
- - Takes any trigger as an argument and forces the triggered window elements to be - "purged" (discarded) after triggering. - -{% highlight java %} -windowedStream.trigger(PurgingTrigger.of(CountTrigger.of(1000))); + + +
+{% highlight scala %}
+public interface WindowFunction |
-
| Delta trigger | -
- - A window is periodically considered for being fired (every 5000 milliseconds in the example). - A window is actually fired when the value of the last added element exceeds the value of - the first element inserted in the window according to a `DeltaFunction`. - + + + +Here we show an example that uses a `WindowFunction` to count the elements in a window. We do this +because we want to access information about the window itself to emit it along with the count. +This is very inefficient, however, and should be implemented with a +`ReduceFunction` in practice. Below, we will see an example of how a `ReduceFunction` can +be combined with a `WindowFunction` to get both incremental aggregation and the added +information of a `WindowFunction`. + +
+
{% highlight java %}
-windowedStream.trigger(new DeltaTrigger.of(5000.0, new DeltaFunction |
-
| Transformation | -Description | -
|---|---|
| Processing time trigger | -
- - A window is fired when the current processing time exceeds its end-value. - The elements on the triggered window are henceforth discarded. - -{% highlight scala %} -windowedStream.trigger(ProcessingTimeTrigger.create); -{% endhighlight %} - |
-
| Watermark trigger | -
- - A window is fired when a watermark with value that exceeds the window's end-value has been received. - The elements on the triggered window are henceforth discarded. - -{% highlight scala %} -windowedStream.trigger(EventTimeTrigger.create); -{% endhighlight %} - |
-
| Continuous processing time trigger | -
- - A window is periodically considered for being fired (every 5 seconds in the example). - The window is actually fired only when the current processing time exceeds its end-value. - The elements on the triggered window are retained. - -{% highlight scala %} -windowedStream.trigger(ContinuousProcessingTimeTrigger.of(Time.seconds(5))); -{% endhighlight %} - |
-
| Continuous watermark time trigger | -
- - A window is periodically considered for being fired (every 5 seconds in the example). - A window is actually fired when a watermark with value that exceeds the window's end-value has been received. - The elements on the triggered window are retained. - -{% highlight scala %} -windowedStream.trigger(ContinuousEventTimeTrigger.of(Time.seconds(5))); -{% endhighlight %} - |
-
| Count trigger | -
- - A window is fired when it has more than a certain number of elements (1000 below). - The elements of the triggered window are retained. - {% highlight scala %} -windowedStream.trigger(CountTrigger.of(1000)); +val input: DataStream[(String, Long)] = ... + +input + .keyBy( |
-
| Purging trigger | -
- - Takes any trigger as an argument and forces the triggered window elements to be - "purged" (discarded) after triggering. - -{% highlight scala %} -windowedStream.trigger(PurgingTrigger.of(CountTrigger.of(1000))); + + + +### WindowFunction with Incremental Aggregation + +A `WindowFunction` can be combined with either a `ReduceFunction` or a `FoldFunction`. When doing +this, the `ReduceFunction`/`FoldFunction` will be used to incrementally aggregate elements as they +arrive while the `WindowFunction` will be provided with the aggregated result when the window is +ready for processing. This allows to get the benefit of incremental window computation and also have +the additional meta information that writing a `WindowFunction` provides. + +This is an example that shows how incremental aggregation functions can be combined with +a `WindowFunction`. + +
+
+{% highlight java %}
+DataStream |
-
| Delta trigger | -
- - A window is periodically considered for being fired (every 5000 milliseconds in the example). - A window is actually fired when the value of the last added element exceeds the value of - the first element inserted in the window according to a `DeltaFunction`. - + + +
{% highlight scala %}
-windowedStream.trigger(DeltaTrigger.of(5000.0, { (old,new) => new - old > 0.01 }))
+val input: DataStream[(String, Long)] = ...
+
+// for folding incremental computation
+input
+ .keyBy( |
-
| Transformation | -Description | -
|---|---|
| Time evictor | -
- - Evict all elements from the beginning of the window, so that elements from end-value - 1 second - until end-value are retained (the resulting window size is 1 second). - - {% highlight java %} -triggeredStream.evictor(TimeEvictor.of(Time.seconds(1))); - {% endhighlight %} - |
-
| Count evictor | -
- - Retain 1000 elements from the end of the window backwards, evicting all others. - - {% highlight java %} -triggeredStream.evictor(CountEvictor.of(1000)); - {% endhighlight %} - |
-
| Delta evictor | -
- - Starting from the beginning of the window, evict elements until an element with - value lower than the value of the last element is found (by a threshold and a - DeltaFunction). - - {% highlight java %} -triggeredStream.evictor(DeltaEvictor.of(5000, new DeltaFunction |
-
| Transformation | -Description | -
|---|---|
| Time evictor | -
- - Evict all elements from the beginning of the window, so that elements from end-value - 1 second - until end-value are retained (the resulting window size is 1 second). - - {% highlight scala %} -triggeredStream.evictor(TimeEvictor.of(Time.seconds(1))); - {% endhighlight %} - |
-
| Count evictor | -
- - Retain 1000 elements from the end of the window backwards, evicting all others. - - {% highlight scala %} -triggeredStream.evictor(CountEvictor.of(1000)); - {% endhighlight %} - |
-
| Delta evictor | -
- - Starting from the beginning of the window, evict elements until an element with - value lower than the value of the last element is found (by a threshold and a - DeltaFunction). - - {% highlight scala %} -windowedStream.evictor(DeltaEvictor.of(5000.0, { (old,new) => new - old > 0.01 })) - {% endhighlight %} - |
-
| Window type | -Definition | -
|---|---|
|
- Tumbling count window - {% highlight java %} -stream.countWindow(1000) - {% endhighlight %} - |
- - {% highlight java %} -stream.window(GlobalWindows.create()) - .trigger(PurgingTrigger.of(CountTrigger.of(size))) - {% endhighlight %} - | -
|
- Sliding count window - {% highlight java %} -stream.countWindow(1000, 100) - {% endhighlight %} - |
- - {% highlight java %} -stream.window(GlobalWindows.create()) - .evictor(CountEvictor.of(1000)) - .trigger(CountTrigger.of(100)) - {% endhighlight %} - | -
|
- Tumbling event time window - {% highlight java %} -stream.timeWindow(Time.seconds(5)) - {% endhighlight %} - |
- - {% highlight java %} -stream.window(TumblingEventTimeWindows.of(Time.seconds(5)) - .trigger(EventTimeTrigger.create()) - {% endhighlight %} - | -
|
- Sliding event time window - {% highlight java %} -stream.timeWindow(Time.seconds(5), Time.seconds(1)) - {% endhighlight %} - |
- - {% highlight java %} -stream.window(SlidingEventTimeWindows.of(Time.seconds(5), Time.seconds(1))) - .trigger(EventTimeTrigger.create()) - {% endhighlight %} - | -
|
- Tumbling processing time window - {% highlight java %} -stream.timeWindow(Time.seconds(5)) - {% endhighlight %} - |
- - {% highlight java %} -stream.window(TumblingProcessingTimeWindows.of(Time.seconds(5)) - .trigger(ProcessingTimeTrigger.create()) - {% endhighlight %} - | -
|
- Sliding processing time window - {% highlight java %} -stream.timeWindow(Time.seconds(5), Time.seconds(1)) - {% endhighlight %} - |
- - {% highlight java %} -stream.window(SlidingProcessingTimeWindows.of(Time.seconds(5), Time.seconds(1))) - .trigger(ProcessingTimeTrigger.create()) - {% endhighlight %} - | -
| Transformation | -Description | -
|---|---|
| Tumbling time window all DataStream → WindowedStream |
-
- - Defines a window of 5 seconds, that "tumbles". This means that elements are - grouped according to their timestamp in groups of 5 second duration, and every element belongs to exactly one window. - The notion of time used is controlled by the StreamExecutionEnvironment. - {% highlight java %} -nonKeyedStream.timeWindowAll(Time.seconds(5)); - {% endhighlight %} - - |
-
| Sliding time window all DataStream → WindowedStream |
-
- - Defines a window of 5 seconds, that "slides" by 1 second. This means that elements are - grouped according to their timestamp in groups of 5 second duration, and elements can belong to more than - one window (since windows overlap by at least 4 seconds) - The notion of time used is controlled by the StreamExecutionEnvironment. - {% highlight java %} -nonKeyedStream.timeWindowAll(Time.seconds(5), Time.seconds(1)); - {% endhighlight %} - - |
-
| Tumbling count window all DataStream → WindowedStream |
-
- - Defines a window of 1000 elements, that "tumbles". This means that elements are - grouped according to their arrival time (equivalent to processing time) in groups of 1000 elements, - and every element belongs to exactly one window. - {% highlight java %} -nonKeyedStream.countWindowAll(1000) - {% endhighlight %} - - |
-
| Sliding count window all DataStream → WindowedStream |
-
- - Defines a window of 1000 elements, that "slides" every 100 elements. This means that elements are - grouped according to their arrival time (equivalent to processing time) in groups of 1000 elements, - and every element can belong to more than one window (as windows overlap by at least 900 elements). - {% highlight java %} -nonKeyedStream.countWindowAll(1000, 100) - {% endhighlight %} - - |
-
| Transformation | -Description | -
|---|---|
| Tumbling time window all DataStream → WindowedStream |
-
- - Defines a window of 5 seconds, that "tumbles". This means that elements are - grouped according to their timestamp in groups of 5 second duration, and every element belongs to exactly one window. - The notion of time used is controlled by the StreamExecutionEnvironment. - {% highlight scala %} -nonKeyedStream.timeWindowAll(Time.seconds(5)); - {% endhighlight %} - - |
-
| Sliding time window all DataStream → WindowedStream |
-
- - Defines a window of 5 seconds, that "slides" by 1 second. This means that elements are - grouped according to their timestamp in groups of 5 second duration, and elements can belong to more than - one window (since windows overlap by at least 4 seconds) - The notion of time used is controlled by the StreamExecutionEnvironment. - {% highlight scala %} -nonKeyedStream.timeWindowAll(Time.seconds(5), Time.seconds(1)); - {% endhighlight %} - - |
-
| Tumbling count window all DataStream → WindowedStream |
-
- - Defines a window of 1000 elements, that "tumbles". This means that elements are - grouped according to their arrival time (equivalent to processing time) in groups of 1000 elements, - and every element belongs to exactly one window. - {% highlight scala %} -nonKeyedStream.countWindowAll(1000) - {% endhighlight %} - - |
-
| Sliding count window all DataStream → WindowedStream |
-
- - Defines a window of 1000 elements, that "slides" every 100 elements. This means that elements are - grouped according to their arrival time (equivalent to processing time) in groups of 1000 elements, - and every element can belong to more than one window (as windows overlap by at least 900 elements). - {% highlight scala %} -nonKeyedStream.countWindowAll(1000, 100) - {% endhighlight %} - - |
-