aggregation operator is an alternative implementation of Aggregate (and Group) operator. In particular for sliding time windows with advance of 1 this operator is faster than the implementation with partial aggregates.
Differences in the use of this operator compared to Aggregate (and Group) operator:
- This operator has a more flexible PQL interface that allows to specify key value parameters.
- This operator does not set end timestamps of the resulting data stream elements. If you need the validity of the aggregation value you need to append an element window of size 1.
- This operator outputs "empty aggregations" if no tuple is valid at a specific point in time. E.g., the sum aggregation function would output 0. This is necessary to determine the end timestamp with a subsequent element window.
These aggregation functions are still in development. Especially the keys for the parameters are preliminary and subject to change.
group_by:An optional list of attributes over which the grouping should occur.
aggregations:A list of aggregate functions (see below).
SUPPRESS_FULL_META_DATA_HANDLING:Boolean flag set to true if the handling of meta data other than Time Interval (e.g. Latency) should be supressed.
The following optional boolean parameters control when a new aggregation value is transferred (see below for useful examples):
eval_at_new_element: Outputs an updated aggregation value when a new element gets valid. In the case that more than one element gets valid at the same time (same start timestamp), this operator outputs for each element an output value in the order of arrival. The default value is
eval_at_outdating: Outputs an updated aggregation value when one ore more elements gets invalid with the value after the removal of the invalid elements. The default value is
eval_before_remove_outdating: Outputs an updated aggregation value before removing the invalid elements instead of after removal. The default value is
eval_at_done: Outputs the value at the time the operator gets the done signal. The default value is
output_only_changes: Suppresses elements that are equal to the previous outputted element. The default value is
false. If you want to use this, make sure the equals-method for every attribute type is implemented.
|Count||Outputs the number of steam elements.|
['FUNCTION' = 'Count']
['FUNCTION' = 'Count', 'OUTPUT_ATTRIBUTES' = 'number_of_elements']
|Sum||Outputs the sum of elements.|
['FUNCTION' = 'Sum']
['FUNCTION' = 'Sum', 'INPUT_ATTRIBUTES' = 'value1']
['FUNCTION' = 'Sum', 'INPUT_ATTRIBUTES' = ['value1', 'value2']]
|Avg||Average value (mean)||TODO (similar to Sum)|
|Min||Min value||TODO (similar to Sum)|
|Max||Max value||TODO (similar to Sum)|
|First||The first element of a window. See example below.|
|You should use the following settings:|
output_only_changes = true
This results in getting the first element in each window. Especially useful with a tumbling window.
|Last||The last element of a window. See example below.|
|You should use the following settings:|
EVAL_AT_NEW_ELEMENT = false
This results in getting the last element in each window. Especially useful with a tumbling window.
|Trigger||The tuple that triggers the output.||TODO|
|Variance||Calculates the variance||TODO (similar to Sum)|
|TopK||Calculates the top-K list||TODO|
|Nest||Nests the valid elements as list.||TODO|
You can use more than one aggregation function:
Here, we use a tumbling window with the "First" aggregate function to only get the first element per 5-minute window.
Here, we use a tumbling window and the "Last" aggregate function to only get the last element per 5-minute window.
Changing the way this operator outputs values
By using the default values, this operator act as Aggregate (and Group) operator (with the limitations explained above). Useful alternative settings are:
trueand add a preceding window with advance.
Remark: In this case, the starttimestamp of the output gets the timestamp of the value, that triggers the output (i.e. the element that states, that the current elements are outdated).
The following example calculates the number of elements in the stream impressions in one minute. It outputs the total number at the end of each minute instead of each update when a new item arrives.