Tumbling and sliding windows

Rather than keeping all the tuples ever inserted, windows are configured to evict expired tuples. In this respect, tumbling windows operate in batches. When a tumbling window fills up, all the tuples in the window are evicted. This process is called a window flush.

Conversely, sliding windows operate in an incremental fashion. When a sliding window fills up, the future tuple insertions result in evicting the oldest tuples in the window.

The details of tuple eviction are defined by the eviction policy. There are different eviction policies. Consider a count-based eviction policy of size c. A tumbling window that is configured in this way flushes whenever the number of tuples in the window reaches c. A similarly configured sliding window will evict the first tuple when c+1^th tuple is received and each new tuple after that evicts the oldest tuple in the window.

The trigger policy

Another important way in which tumbling and sliding windows differ is with how they trigger. Window triggers are events that indicate that the window is ready for processing. What the term processing means is application specific. In the context of an Aggregate operator, it might be the computation of an aggregation function, for a Sort operator it might be the sorting of the window. The window trigger event is the same as the window flush event for tumbling windows.

For a sliding window, the trigger policy defines when the window triggers. Consider the simple case of a count-based trigger policy of size c. The window will trigger after c new tuples are received since the last trigger.

Partitioned variants of sliding and tumbling windows

The SPL language also supports partitioned variants of sliding and tumbling windows. For partitioned windows, eviction and trigger policies apply to individual partitions, independent of other partitions. Each tuple that is inserted into the window is associated with a user-specified partition. In effect, each partition has its own subwindow and all the windowing configurations apply to the subwindows, independently. As a result, window events such as evictions and triggers have partitions that are associated with them.

For tumbling windows, the subwindows that are associated with the partitions are removed when the windows are flushed. Conversely, for sliding windows the subwindows that are associated with the partitions might never become empty in certain cases. For example, consider a count or delta-based eviction policy; when the window is already partially populated, it can never become empty. It is important to understand this distinction, as it has implications on the memory usage.

Eviction and trigger policies

Partition eviction policies

Partitioned windows are used to group related tuples for processing. Each partition contains 0 to n tuples. In partitioned windows, tuples might never be evicted from a partition. For example, consider an Aggregate operator with a partitioned tumbling window that has a trigger policy of count(100). If the application generates only 99 tuples in each partition, the partitions would never trigger, and the memory usage of the window would continue to increase. Sometimes, the application requirements specify that if no tuples are generated during a specific duration, the existing tuples in the window are not useful. In such situations, you must delete partitions and recover memory that is allocated to the tuples.

Execution order of events

The executions of the window events are set off by the insert calls made on the window object. The insertion, eviction, and trigger events are run in a certain order when the insert call is made. As an exception, for time-based window policies, the eviction and trigger processing might proceed independently of the insertion, depending on the specific time-based policies used. The specific order that is used for delivering window events depends on the window policies specified. Two tables, one for tumbling windows (Table 1) and another for sliding windows (Table 2) summarize the execution order of the window events for different eviction and trigger policies.

Notice the order of insertion and eviction are different for count-based and delta-based eviction policies. This difference is because for a count-based tumbling window of size n, the window is full when the n^th tuple is received. Whereas for a delta-based window of size delta-n, the window is full when a tuple that increases the delta value beyond n is received. In the latter case, the tuples are first evicted from the window and then the newly received tuple is inserted into the emptied window.

For sliding windows that do not involve time-based policies, eviction always precedes insertion, since evictions are needed to make room for the new tuple that arrives. However, the trigger processing might come first or last depending on whether there are count-based or delta-based trigger policies. For count-based trigger policies, the triggering happens at the end because the newly inserted tuple is part of the contents of the window that needs to be triggered. For instance, for a count-based trigger policy of size n, the n^th tuple that is received since the last trigger results in a new trigger and that n^th tuple is part of the triggered window, meaning that the trigger succeeds the insertion. However, the delta-based windows are different because the newly received tuple is not part of the contents of the window that needs to be triggered. For instance, for a delta-based trigger policy of size delta-n, the tuple that increases the delta value beyond n since the last trigger results in a new trigger and that tuple is not part of the triggered window, meaning that the trigger precedes both the insertion and the eviction. For time-based eviction policies, the eviction happens independently of other events, and similarly, for time-based trigger policies, the trigger happens independently of other events.

Defining tumbling and sliding windows

template <class TupleType,
          class PartitionType=int32_t,
          class DataType=std::deque<TupleType>,
          class StorageType=std::tr1::
            unordered_map<PartitionType,DataType>  >
class Window;

The types that are involved in the template are:

• TupleType is the type of items that are stored in the window. It can be a pointer or non-pointer type.
• PartitionType is the type of the window partitioning attribute.
• DataType is the type of the structure that holds the tuples for each partition.
• StorageType is the type of the structure that holds the data that is associated with all the partitions. It is a mapping from the partition attribute to the data for that partition.

All of these types are also defined within the Window, TumblingWindow, and SlidingWindow classes as typedefs. In general, the developers customize only TupleType and PartitionType. The following are some example templates.

class TumblingWindow<TType>;
class SlidingWindow<TType *, float32>; // partitioned window

For windows that store pointers to tuples, the memory management responsibilities belong to the user.

Tumbling window constructors take three parameters, a reference to the operator that contains the window, an index to the input port that this window is associated with, and the eviction policy for the window. Sliding windows are similar, but they also take the trigger policy, as the last parameter. The constructor signatures are given as follows:

TumblingWindow(Operator & oper, uint32_t port,
               WindowPolicy const & evictionPolicy);
SlidingWindow(Operator & oper, uint32_t port,
              WindowPolicy const & evictionPolicy,
              WindowPolicy const & triggerPolicy);

The following are some example window configurations:

CountWindowPolicy pC5(5), pC2(2); // count-based policies
PunctWindowPolicy pP; // punctuation-based policy
DeltaWindowPolicy<TT,TT::age_type,&TT::get_age> pDage2(2), pDage0(0); // delta-based policy
DeltaWindowPolicy<TT,TT::time_type,&TT::get_time> pDtime2(2.0); // delta-based policy
TimeWindowPolicy pT8(8.0), pT2(2.0); // time-based policy

TumblingWindow<TT> wT_C5(op, 0, pC5); // tumbling, count(5)
TumblingWindow<TT> wT_P(op, 0, pP); // tumbling, punct()
TumblingWindow<TT> wT_Dage2(op, 0, pDage2); // tumbling, delta(age,2)
TumblingWindow<TT> wT_T8(op, pT8); // tumbling, time(8)

SlidingWindow<TT> wS_C5_C2(op, 0, pC5, pC2); // sliding, count(5), count(2)
SlidingWindow<TT> wS_C5_Dage2(op, 0, pC5, pDage2); // sliding, count(5), delta(age, 2)
SlidingWindow<TT> wS_Dage2_C5(op, 0, pDage2, pC5); // sliding, delta(age, 2), count(5)
SlidingWindow<TT> wS_Dage2_Dage0(op, 0, pDage2, pDage0); // sliding, delta(age, 2), delta(age, 0)
SlidingWindow<TT> wS_Dage2_T2(op, 0, pDage2, pT2); // sliding, delta(age, 2), time(2)
SlidingWindow<TT> wS_T8_C5(op, 0, pT8, pC5); // sliding, time(8), count(5)
SlidingWindow<TT> wS_T8_Dage2(op, 0, pT8, pDage2); // sliding, time(8), delta(age, 2)
SlidingWindow<TT> wS_T8_T2(op, 0, pT8, pT2); // sliding, time(8), time(2)

Inserting tuples

SPL run time has an event-driven windowing API. Except for certain events that are fired based on time-based eviction and trigger polices, most window events fire as a result of tuples or punctuation being inserted into the window. The following are the signatures of the two window functions available for inserting tuples and punctuation into the window.

void insert(TupleType const & tuple,
            PartitionType const & partition=PartitionType());
void insert(Punctuation const & punct);

The partition parameter of the tuple insertion function is optional. For non-partitioned windows, it can be omitted. A non-partitioned window has a single default partition with a default partition attribute type (int32_t) and value (0).

Evicting partitions

If a partition eviction policy is defined for a window, each time a tuple is inserted into a partition, the window is checked to see whether partition eviction is required. If partition eviction is required and the LRU option is selected, the default onWindowPartitionEvictionSelection event handler is called to select the partitions to evict. The oldest partitions are selected for eviction from a time-ordered list of partitions. If partition eviction is required and the OperatorDefined option is selected, the user-defined onWindowPartitionEvictionSelection event handler is called to allow the operator to select partitions for eviction.

If partitions are selected for eviction and the onwindowPartitionEvicition event handler is registered, it is called to take user-defined actions before it deletes the partitions. For example, you can delete the tuples in a partition to recover the memory that is allocated to the tuples before you delete the partitions.

Tumbling window summarizer

A tumbling window summarizer can be used to remove the need for the Windowing library to retain all tuples in a tumbling window. If your operator would like to use the facilities of the windowing library to group and partition tuples, but just needs to look at incoming tuples once, a tumbling window summarizer can be used. An example is the Sum() function in Aggregate for a tumbling window.

Window events

Various events that are fired by windows are captured through the WindowEvent template class. WindowEvent has the same template form as the window classes themselves. Window classes also include a typedef that is called WindowEventType that identifies the window event class for the particular window. To handle window events, register a class that derives from WindowEvent with the window at hand. Two events are common to both tumbling and sliding windows. They are listed here as they appear in the WindowEvent class:

virtual void beforeTupleInsertionEvent(
    WindowType & window, TupleType const & tuple, PartitionType const & partition) {}
virtual void afterTupleInsertionEvent(
    WindowType & window, TupleType & tuple, PartitionType const & partition) {}

beforeTupleInsertionEvent is fired right before a tuple is inserted into the window, whereas afterTupleInsertionEvent is fired right after a tuple is inserted into the window.

Three events are specific to the tumbling windows. They are listed here as they appear in the WindowEvent class:

virtual void beforeWindowFlushEvent(WindowType & window, PartitionType const & partition) {}
virtual void afterWindowFlushEvent(WindowType & window, PartitionType const & partition) {}
virtual void onEmptyWindowPunctEvent(WindowType & window) {}

beforeWindowFlushEvent is fired right before a tumbling window is flushed, whereas afterWindowFlushEvent is fired right after a tumbling window is flushed. onEmptyWindowPunctEvent is fired when a tumbling window has trigger punct, and a punctuation is received when there are no tuples in the window. The operator detects that this situation occurs and takes an action (such as submitting a Window punctuation).

Four events are specific to the sliding windows. They are listed here as they appear in the WindowEvent class:

virtual void beforeTupleEvictionEvent(
     WindowType & window, TupleType & tuple, PartitionType const & partition) {}
virtual void afterTupleEvictionEvent(
    WindowType & window, TupleType & tuple, PartitionType const & partition) {}
virtual void onWindowTriggerEvent(
    WindowType & window, PartitionType const & partition) {}
virtual void onWindowInitialFullEvent(
    WindowType & window, PartitionType const & partition) {}

beforeTupleEvictionEvent is fired right before a tuple is evicted from a sliding window, whereas afterTupleEvictionEvent is fired right after a tuple is evicted from a sliding window. onWindowTriggerEvent is fired when a sliding window is triggered. Finally, onWindowInitialFullEvent is fired when a sliding window is full for the first time. This function is particularly useful when the window trigger events are to be ignored before the window becomes full for the first time (a common use case for operators like Aggregate).

Accessing window data

There are three functions available for accessing data that is stored within windows. These functions are defined in the Window class, as follows:

StorageType & getWindowStorage();
DataType & getWindowData(PartitionType const & partition);
DataType & getWindowData();

The getWindowStorage function returns a reference to the main window storage area, which maps each partition to its own specific storage area that contains window data. For a partition, its window data can be accessed through the getWindowData function. Also, a version of getWindowData exists that does not take a partition parameter. This version is used for non-partitioned windows and creates the default partition if it does not already exist.

Window events do not involve multi-threading in the absence of time-based eviction and trigger policies. Otherwise, each time-based policy of a window brings in a thread of its own. For instance, a tumbling window introduces a thread when it is configured with a time-based eviction policy. A sliding window can bring in up to two threads, as it might have a time-based trigger policy in addition to a time-based eviction policy. For a window with a time-based policy, the window is always locked when an event is fired, so the users do not need to deal with locking inside event handlers. In a time-based eviction policy, the tuple eviction and window flush events (before/afterTupleEvictionEvent and before/afterWindowFlushEvent) are fired by the eviction thread. In a time-based trigger policy, the window trigger events (onWindowTriggerEvent) are fired by the trigger thread.

To access the window contents from outside an event handler, in the presence of time-based eviction or trigger policies, the AutoWindowDataAcquirer class can be used. The usage is as follows:

{ AutoWindowDataAcquirer awa(window);
  ... // access the window data safely
}

The AutoWindowDataAcquirer object, when constructed, calls acquireData() on the window, and when destructed (when it goes out of scope in the example), it calls releaseData() on the window. These operations reduce to a no-op for windows that are free of time-based eviction and trigger policies. As a result, this pattern can be safely used in the general case without worrying about unnecessary processing.

Putting it all together

To summarize, here is a list of the steps that are involved in using the windowing library from within a primitive operator:

• Add member variables to your operator that represent the windows
• Extend your class from WindowEvent and add relevant member function overrides to handle window events
• Add initialization code to your operator's constructor to set up the windowing policies for your windows
• Add registration code to your operator's constructor to register the events of interest
• Add logic to your processing functions to insert items into the window
• Add logic to your event handler function to implement windowing functionality

Look at an end-to-end example of using the windowing library in a non-generic operator. This example is extended to a generic operator that handles various types of windows in Window handling.

For illustration purposes, use a small application called SensorQuery. This application aims at answering queries that are posed against the recent history of sensor readings. It contains two sources, one generating queries, and another generating sensor readings. Create an operator that keeps the last n sensor readings and evaluates the incoming queries against these readings, outputting any matches that are found. The matching is based on the distance of the query to the sensor that originates the reading. In other words, this operator keeps a sliding window with a count-based eviction policy of n on the sensor readings, and matches the incoming queries against that window, using nested loop processing. To make things a little more interesting, this operator also tracks how many times a sensor reading was matched during its stay within the window, and outputs readings that are popular when such readings are evicted from the window. A separate output stream is used for this purpose.