The 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 then the implementation with partial aggregates.
Differences in the use of this operator compared to Aggregate (and Group) operator:
This 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).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 true
.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 true
.eval_before_remove_outdating
: Outputs an updated aggregation value before removing the invalid elements instead of after removal. The default value is false
.eval_at_done
: Outputs the value at the time the operator gets the done signal. The default value is false
.output_only_changes
: Suppresses elements that are equal to the previous outputted element. The default value is false
.Function Name | Description | Parameters | Examples | ||||||||||||
---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
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']] |
counted = AGGREGATION({AGGREGATIONS = [['FUNCTION' = 'Count']], GROUP_BY = ['publisher', 'item']}, windowed) |
You can use more than one aggregation function:
counted = AGGREGATION({AGGREGATIONS = [['FUNCTION' = 'Count'], ['FUNCTION' = 'Sum', 'INPUT_ATTRIBUTES' = 'value1']], GROUP_BY = ['publisher', 'item']}, windowed) |
By using the default values, this operator act as Aggregate (and Group) operator (with the limitations explained above). Useful alternative settings are:
eval_at_new_element
to false
and eval_before_remove_outdating
to true
and add a preceding window with advance: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.
windowed = TIMEWINDOW({size = [1, 'Minutes'], ADVANCE = [1, 'MINUTES']}, impressions) impressions_per_minute = AGGREGATION({AGGREGATIONS = [['FUNCTION' = 'Count']], EVAL_AT_NEW_ELEMENT = false, EVAL_BEFORE_REMOVE_OUTDATING = true}, windowed) |