Configure data aggregation
A data aggregation node aggregates messages from the data parsing node within a specified time window, similar to a Flink SQL window function. You can generate diverse aggregated data for downstream analysis or output.
Usage notes
The only supported Flink SQL window function is the tumbling window (TUMBLE). For more information, see Tumbling windows.
The data aggregation node supports only the following fixed-length time windows: 10 s, 15 s, 30 s, 1 min, 5 min, 15 min, and 30 min.
-
For time windows longer than 30 minutes, use the SQL analysis workbench to schedule hourly jobs. For more information, see SQL analysis.
-
If you have custom requirements, contact the helpdesk.
Scenarios
In a campus energy-saving system, you need to identify the conference room with the highest air conditioner temperature (temperature) each minute. Define a data aggregation node with the aggregation granularity set to the conference room ID (roomId), the window length set to 1 minute, and the aggregation algorithm set to MAX. The output fields are the conference room ID (roomId) and the maximum temperature per minute (max_temperature).
Prerequisites
Ensure that data calculation expressions or data filtering filters are set up. For more information, see Configure data computing and data filtering.
Background information
-
For an overview of the data parsing feature, see Usage Notes.
-
For instructions on using the data parsing workspace, see Data Parsing Workspace Instructions.
Procedure
-
In the middle canvas, click the add
icon following the current node. -
In the node list that appears, click the Data Aggregation node.
-
On the canvas, click the Data Aggregation node. In the configuration pane on the right, configure the fields as described in the following table.
Configuration item
Parameter
Description
Example
Basic configuration
Grouping field
Select the fields for data partitioning in the window function (aggregation granularity). These field values are not affected by the window aggregation logic. Examples: ProductKey and DeviceName.
To find the maximum temperature per conference room, select the conference room ID (roomId) as the grouping field.
Window length
Select a window length. Valid values: 10s, 15s, 30s, 1min, 5min, 15min, and 30min.
-
For windows longer than 1 hour, use hourly scheduling in the SQL workbench. For more information, see Set a task scheduling policy in SQL analysis.
-
If you have custom requirements, contact the helpdesk.
In the Scenarios example: 1min.
List of fields to be aggregated
Set the following parameters:
-
Field to be aggregated: Select a numeric field whose values are used for the aggregate operation.
-
Result field name: Can contain only digits, letters, and underscores (_). It cannot start with a digit or duplicate an existing field name. Maximum length: 30 characters.
In the Scenarios example:
-
Field to be aggregated: Select the air conditioner temperature temperature.
-
Result field name: The maximum temperature per minute, max_temperature.
-
Aggregate operation: Select MAX.
Advanced configuration
Delay Tolerance
Unit: seconds.
If messages arrive out of order and are reported after their actual window closes, set this parameter to ensure that late data is calculated in the correct window. This is similar to the Flink SQL watermark mechanism. For more information, see Time attributes in Flink SQL.
Use the default value.
The following is a sample configuration for a specific application scenario:
-
-
In the upper-right corner of the data parsing workbench, click Save.
ImportantThe output of this node includes only the grouping fields and the aggregation result fields. Other fields from the preceding node are not passed to the output.
What to do next
After you configure the data aggregation node, continue configuring other processing nodes or the destination node to complete the parsing task.