Aggregation Connector

Overview

The Aggregation connector buffers incoming payloads in a window and emits a single aggregated payload per window flush. It supports time-based and count-based windows with multiple aggregation functions per field.

Connector Type: MeddleAggregation

Tumbling Window

Fixed, non-overlapping time windows. The buffer is fully drained at every flush.

{
  "type": "MeddleAggregation",
  "config": {
    "windowType": "tumbling",
    "windowSize": 60000,
    "fields": [
      { "key": "temperature", "function": "avg", "outputKey": "temperature_avg" },
      { "key": "temperature", "function": "max", "outputKey": "temperature_max" }
    ]
  }
}

Every 60 seconds, emits the average and maximum of temperature collected during the window, then resets the buffer.

Sliding Window

Overlapping time windows. After each flush, the second half of the buffer is retained so the next emission overlaps with the previous one.

{
  "type": "MeddleAggregation",
  "config": {
    "windowType": "sliding",
    "windowSize": 60000,
    "slideSize": 30000,
    "fields": [
      { "key": "pressure", "function": "avg", "outputKey": "pressure_avg" }
    ]
  }
}

Useful for smoother moving averages where each emission shares half its samples with the previous window.

Count Window

Flushes as soon as the buffer contains countSize payloads, regardless of elapsed time.

{
  "type": "MeddleAggregation",
  "config": {
    "windowType": "count",
    "countSize": 100,
    "fields": [
      { "key": "vibration", "function": "max", "outputKey": "vibration_peak" },
      { "key": "vibration", "function": "count", "outputKey": "samples" }
    ]
  }
}

Emits one payload for every 100 samples received.

Configuration Parameters

• windowType (required): tumbling, sliding, or count
• windowSize: Window duration in milliseconds (required for tumbling and sliding)
• slideSize: Slide interval in milliseconds (used with sliding)
• countSize: Number of samples per window (required for count)
• fields (required, min 1): Array of aggregation field definitions
• groupByKeys: Optional list of payload keys to group aggregations by

Field Definition

• key: Input payload key to read from
• function: One of sum, avg, min, max, count, first, last
• outputKey: Key under which the aggregated value is written

Aggregation Functions

• sum — Sum of numeric values in the window
• avg — Arithmetic mean of numeric values
• min / max — Minimum / maximum numeric value
• count — Number of payloads in the window
• first / last — First / last value observed for the key (any type)

Non-numeric values are skipped for numeric functions and reported as type-mismatch errors on the error channel.

Use Cases

1. Smoothing noisy sensor readings with a sliding average
2. Per-minute or per-hour KPIs for dashboards
3. Batching events before writing to a database or message broker
4. Peak detection over a fixed sample count

Example: Multi-Field Tumbling Window

{
  "type": "MeddleAggregation",
  "config": {
    "windowType": "tumbling",
    "windowSize": 30000,
    "fields": [
      { "key": "temperature", "function": "avg", "outputKey": "temperature_avg" },
      { "key": "temperature", "function": "min", "outputKey": "temperature_min" },
      { "key": "temperature", "function": "max", "outputKey": "temperature_max" },
      { "key": "status", "function": "last", "outputKey": "status_last" }
    ]
  }
}

Best Practices

• Choose tumbling for clean periodic reports; sliding for smoother trends
• Use count windows for high-frequency streams where time is not the natural unit
• Keep windows small enough to bound memory but large enough to be statistically meaningful
• Use first / last to carry non-numeric context (e.g. machine state) alongside numeric aggregates

Related Connectors

• Trigger - Conditional logic on aggregated values
• Anomaly Detection - Detect outliers
• Chart - Visualize aggregated series