One important operation in streaming and CEP applications is the aggregation. We will give an example on how Odysseus computes aggregations. A problem for aggregations in combination with sliding windows is how to handle events which leave a window (they become invalid). A simple approach is to keep every event and calculate the aggregation on evaluation time. Here, the event that get invalid can be simply removed. But this causes a large memory overhead. Odysseus adapts the concept of online aggregations. In this concept, only the minimal needed information is kept to calculate an aggregation over a window by utilizing so called partial aggregates. A very intuitive example is the calculation of an average: The partial aggregate needs to keep a running sum and the count of events aggregated so far. This can be used to calculate the concrete result anytime by the division of sum and count.

But with this partial aggregates, it is not possible to remove the values of the first event without knowing its content. In Odysseus we adapted the approach of Partial aggregates. Partial aggregates are built only from events that overlap in their time interval. For this, new events must be combined with different existing partial aggregates.

 

 

Fig. 2 illustrates an example. To make the gure more readable,

this example shows a count aggregation. The AGGRE-

GATE operator keeps a state of all current partial aggregates.

In this example there are two partial aggregates with

This operator is used to aggregate and group the input stream.

Parameter

Aggregation Functions

The set of aggregate functions is extensible. The following list is in the core Odysseus:


Some nonstandard aggregations: These should only be used, if you a familiar with them:

Example:

PQL
output = AGGREGATE({
                    group_by = ['bidder'], 
                    aggregations=[ ['MAX', 'price', 'max_price', 'double'] ]
                   }, input)

// Parital Aggregate example
pa = AGGREGATE({
          name='PRE_AGG',
          aggregations=[
            ['count', 'id', 'count', 'PartialAggregate'],
            ['sum', 'id', 'avgsum', 'PartialAggregate'],
            ['min', 'id', 'min', 'PartialAggregate'],
            ['max', 'id', 'max', 'PartialAggregate']
          ],
          outputpa='true'        
        },
        nexmark:person
      )
      
out = AGGREGATE({
          name='AGG',
          aggregations=[
            ['count', 'count', 'count', 'Integer'],
            ['sum', 'avgsum', 'sum', 'Double'],
            ['avg', 'avgsum', 'avg', 'Double'],
            ['min', 'min', 'min', 'Integer'],
            ['max', 'max', 'max', 'Integer']
          ]        
        },
        pa
      )

/// Example for aggregations on multiple attributes
out = AGGREGATE({
          aggregations=[
            ['corr', ['x', 'y'], 'correlation', 'Double'],
            ['cov', ['x', 'y'], 'covariance', 'Double']
          ]        
        },
        input
      )
CQL
SELECT MAX(price) AS max_price FROM input GROUP BY bidder