Time Window Collector
The unit is a Processor type unit.
The Time Window Collector unit keeps events in memory until a certain time condition is met. It can be configured to either use a global window for all events or to create a new window for each key.
An event comes in through the in port. The event is stored in the window associated with the event key. If the time condition of the window is met, all events are released.
Windowed events are sent to the out output port.
Events that do not match the time condition are sent to the discarded port.
If an error occurs, the input events are enriched with standard error fields and sent to the error port.
After dragging this unit into the Flow canvas, double-click it to access its configuration options. The following table describes the configuration options of this unit:
|Name||Enter a name for the unit. It must start with a letter, and cannot contain spaces. Only letters, numbers, and underscores are allowed.|
|Key field(s)||Enter the name of an input event field(s) containing key(s).|
The size of the window in seconds. Must be greater than 0.
If this number is smaller than the "Purge size", a sliding window is used.
For example, imagine you add a size of 300 seconds (5 minutes) with a timestamp at 11:00. The sliding window will be 11:00 - 11:05. In order to determine the next window, the values set in the Purge size field are used.
If this number is equal to "Purge size", a tumbling window is used.
Amount of seconds after which to purge from the window when it is considered complete.
All events received at the purged seconds are removed from the window.
If this number is larger than "size", a sliding window is used. If this number is equal to "size", a tumbling window is used.
|Timestamp field||The name of an event field that contains the timestamp. If empty, the unit will use the System time to calculate the event timestamp. Changing this field in a running flow diagram produces a reset of the unit. You can also write java class qualified name.|
|Key(s) field||Enter a name for the output event field containing key(s).|
|Window field||Enter a name for the output event field containing the window list.|
|Size field||Enter a name for the output event field containing the window size.|
|Lower limit field||Enter a name for the output event field containing the lower limit of the current window.|
|Upper limit field||Enter a name for the output event field containing the upper limit of the current window.|
|Allow empty collections||When collection occurs in output ports, whether empty collections should emit a result (true) or not (false). This is set to false by default.|
|in||All events enter through this port.|
|out||Outputs events properly stored.|
|discarded||Outputs events that do not match the time condition.|
|error||Signals when an error occurs. Outputs input events enriched with standard error fields.|
In this example, we wish to accumulate events containing a certain IP address for three seconds before releasing them.
For this, we add a Generator unit to generate events from across Devo. Then, we link it to the input port of the Map unit, which will add the key fields, and join its out output port to a Time window collector unit. Finally, we link the out output port to a Devo Sink unit to send it to the configured my.app table.
In the Map unit, configure the Fields to add as follows:
In the Time Window Collector, enter the IP field in the key fields field. To accumulate for 3 seconds, enter 3 in the size field. Enter a purge of 1 to slide the window one second within the 3.
In the Devo Sink unit properties, you must specify the my.app destination.
You can try this flow by downloading this JSON file and uploading it to your domain using the Import option.