Skip to main content

Complete Pipeline

Copy pipeline YAML
Review it for your environment before you deploy.

This pipeline combines all time-windowed aggregation techniques from this tutorial:

  • Tumbling windows - One-minute summaries from the newest interval
  • Sliding windows - Five-minute windows emitted every minute
  • Multi-level aggregation - Combine window types for different use cases

Full Configuration​

Complete pipeline
aggregate-time-windows.yaml
name: aggregate-time-windows-complete
type: pipeline
description: Complete time-windowed aggregation example with tumbling, sliding, and multi-level aggregation patterns.
namespace: production
labels:
  category: data-transformation
  pattern: time-windows

config:
  input:
    kafka:
      addresses:
        - '${KAFKA_BROKER_1:kafka-1:9092}'
        - '${KAFKA_BROKER_2:kafka-2:9092}'
      topics: ['sensor-events']
      consumer_group: 'time-window-aggregation-${INSTANCE_ID:default}'

  pipeline:
    processors:
      - try:
          - mapping: |
              root = if !this.exists("sensor_id") {
                throw("Missing required field: sensor_id")
              } else if !this.exists("timestamp") {
                throw("Missing required field: timestamp")
              } else if !this.exists("temperature") {
                throw("Missing required field: temperature")
              } else {
                this
              }
              root.processing_started_at = now().ts_format("2006-01-02T15:04:05.000Z")
              root.pipeline_instance = "${INSTANCE_ID:default}"
              root.location = this.location.or("unknown_location")
              root.facility = this.facility.or("unknown_facility")
      - catch:
          - log:
              level: WARN
              message: 'input validation failed: ${! error() }'
          - mapping: root = deleted()
      - archive:
          format: json_array
      - mapping: |-
          let end = meta("window_end_timestamp").ts_parse("2006-01-02T15:04:05Z07:00").ts_unix_nano() / 1000000000
          let minute = 60
          root = this.map_each(event -> [
            event.merge({"aggregation_type":"sliding","group_key":event.sensor_id+"|sliding","window_duration_minutes":5}),
            if event.timestamp.ts_parse("2006-01-02T15:04:05Z07:00").ts_unix_nano() / 1000000000 >= $end - $minute { event.merge({"aggregation_type":"tumbling","group_key":event.sensor_id+"|tumbling","window_duration_minutes":1}) },
            if event.timestamp.ts_parse("2006-01-02T15:04:05Z07:00").ts_unix_nano() / 1000000000 >= $end - $minute { event.merge({"aggregation_type":"multi_level","aggregation_level":"sensor","group_key":event.sensor_id+"|sensor","window_duration_minutes":1}) },
            if event.timestamp.ts_parse("2006-01-02T15:04:05Z07:00").ts_unix_nano() / 1000000000 >= $end - $minute { event.merge({"aggregation_type":"multi_level","aggregation_level":"location","group_key":event.location+"|location","window_duration_minutes":1}) }
          ]).flatten().filter(event -> event != null)
      - unarchive:
          format: json_array
      - group_by_value:
          value: ${! this.group_key }
      - mapping: |-
          let events = json().from_all().sort_by(event -> event.timestamp)
          let first = $events.index(0)
          let values = $events.map_each(event -> event.temperature)
          let end = meta("window_end_timestamp").ts_parse("2006-01-02T15:04:05Z07:00").ts_unix_nano() / 1000000000
          let minute = 60
          root = if batch_index() == 0 {
            {
              "aggregation_type": $first.aggregation_type,
              "aggregation_level": $first.aggregation_level,
              "group_key": $first.group_key,
              "aggregation_timestamp": now(),
              "sensor_id": $first.sensor_id,
              "location": $first.location,
              "window_start": ($end - $minute * $first.window_duration_minutes).ts_format("2006-01-02T15:04:05Z07:00"),
              "window_end": meta("window_end_timestamp"),
              "window_duration_minutes": $first.window_duration_minutes,
              "event_count": $events.length(),
              "temperature_avg": ($values.sum() / $values.length() * 100).round() / 100,
              "temperature_min": $values.min(),
              "temperature_max": $values.max(),
              "temperature_change": if $first.aggregation_type == "sliding" { $values.index(-1) - $values.index(0) } else { deleted() },
              "temperature_trend": if $first.aggregation_type == "sliding" {
                if $values.index(-1) - $values.index(0) > 1 { "increasing" }
                else if $values.index(-1) - $values.index(0) < -1 { "decreasing" }
                else { "stable" }
              } else { deleted() },
              "sensor_count": $events.map_each(event -> event.sensor_id).unique().length(),
              "location_count": $events.map_each(event -> event.location).unique().length(),
              "data_quality_score": if $events.length() >= 50 { 0.95 }
                else if $events.length() >= 30 { 0.85 }
                else if $events.length() >= 10 { 0.70 }
                else { 0.50 }
            }
          } else { deleted() }
      - mapping: 'root = {"aggregation": this}.merge(this)'

  output:
    broker:
      pattern: fan_out
      outputs:
        - http_client:
            url: '${ANALYTICS_ENDPOINT:http://localhost:8080}/aggregations'
            verb: POST
            headers:
              Content-Type: application/json
            batching:
              count: 500
              period: '10s'
            timeout: '15s'
            retries: 2

        - file:
            path: '/var/expanso/buffer/aggregations-${!timestamp_unix()}.jsonl'
            codec: lines
  buffer:
    system_window:
      size: 5m
      slide: 1m
      timestamp_mapping: root = this.timestamp

logger:
  level: '${LOG_LEVEL:INFO}'
  format: json

metrics:
  prometheus:
    path: '/metrics'

Input and window boundary​

The displayed configuration consumes the sensor-events Kafka topic. Each record needs sensor_id, timestamp, and numeric temperature; optional location and facility fields support rollups. The system_window buffer accumulates events before grouping by value and computing batch statistics. It does not implement inactivity-based session windows.

The download below is the simpler one-minute HTTP aggregation variant at /sensors/readings, rather than the Kafka configuration displayed here.

Deploy​

# Deploy to Expanso orchestrator
expanso-cli job deploy aggregate-time-windows.yaml

# Or run locally with expanso-edge
expanso-edge run --config aggregate-time-windows.yaml

Download​

Download aggregate-time-windows.yaml

What's Next?​