data-streaming-aggregations

name	data-streaming-aggregations
description	Windowing, sessionization, time-series aggregation, and late data handling for streaming systems

Streaming Aggregations

Scope: Windowing strategies, sessionization, time-series aggregation, watermarks, late data handling Lines: 385 Last Updated: 2025-10-27 Format Version: 1.0 (Atomic)

When to Use This Skill

Use this skill when:

• Aggregating unbounded streams (counts, sums, averages)
• Implementing time-based windows (tumbling, sliding, session)
• Building real-time dashboards and metrics
• Detecting patterns in time-series data
• Handling out-of-order and late-arriving data
• Computing sessionized analytics (user sessions, click streams)
• Building real-time anomaly detection
• Creating streaming leaderboards and rankings

Core Concepts

Window Types

Tumbling Windows
  → Fixed size, non-overlapping
  → Example: Count events per 5 minutes
  → Use: Periodic reports, batched processing
  → Memory: O(window_size)

Sliding Windows
  → Fixed size, overlapping
  → Example: Moving average over last 10 minutes
  → Use: Continuous metrics, smoothed trends
  → Memory: O(window_size * slide_factor)

Session Windows
  → Dynamic size, gap-based
  → Example: User sessions with 30-min inactivity
  → Use: User behavior, conversation threads
  → Memory: O(active_sessions)

Global Windows
  → Unbounded, single window
  → Example: All-time counts
  → Use: Stateful processing without time bounds
  → Memory: O(cardinality)

Time Semantics

Event Time
  → Time when event occurred
  → Requires: Timestamps in data
  → Accurate but complex (late data)
  → Use: Financial, billing, analytics

Processing Time
  → Time when event processed
  → Simple, low latency
  → Inaccurate for time-based logic
  → Use: Monitoring, system metrics

Ingestion Time
  → Time when event entered system
  → Middle ground
  → Use: Approximation when no event time

Watermarks

Watermark(T)
  → "All events before T have arrived"
  → Heuristic, not guarantee
  → Triggers window computation

Strategies:
  → Perfect: Wait forever (impractical)
  → Bounded delay: T = max_timestamp - delay
  → Percentile: Allow X% late data
  → Punctuation: Explicit markers in stream

Late Data Handling

Strategies:
  → Drop: Ignore late data (simplest)
  → Update: Recompute window (expensive)
  → Side output: Route to separate stream
  → Allowed lateness: Accept within window

Trade-offs:
  → Accuracy vs Latency
  → Completeness vs Timeliness

Patterns

Pattern 1: Tumbling Window Aggregation (Rust/Timely)

use timely::dataflow::operators::{ToStream, Map, Inspect};
use differential_dataflow::input::Input;
use differential_dataflow::operators::{Reduce, Consolidate};
use differential_dataflow::operators::arrange::ArrangeByKey;

#[derive(Clone, Debug, Hash, PartialEq, Eq, PartialOrd, Ord)]
struct Event {
    user_id: u32,
    value: i32,
    timestamp: u64,  // Event time in milliseconds
}

fn main() {
    timely::execute_from_args(std::env::args(), |worker| {
        let mut input = worker.dataflow::<u64, _, _>(|scope| {
            let (input, events) = scope.new_collection();

            // Tumbling window: 5-minute (300000ms) windows
            let window_size = 300_000u64;

            events
                .map(move |event| {
                    // Assign to window based on event time
                    let window_id = event.timestamp / window_size;
                    ((event.user_id, window_id), event.value)
                })
                .reduce(|_key, values, output| {
                    let sum: i32 = values.iter()
                        .map(|(value, diff)| value * diff)
                        .sum();
                    let count:  = values.()
                        .(|(_, diff)| diff)
                        .();
                    output.(((sum, count), ));
                })
                .(|((user_id, window_id), (sum, count))| {
                    (,
                             user_id, window_id, sum, count);
                });

            input
        });

        
        input.(Event { user_id: , value: , timestamp:  });
        input.(Event { user_id: , value: , timestamp:  });
        input.(Event { user_id: , value: , timestamp:  }); 
        input.();
        worker.(|| input.().(&));
    }).();
}

Pattern 2: Sliding Window (Python)

import time
from collections import deque
from dataclasses import dataclass
from typing import Deque, Dict
import threading

@dataclass
class Event:
    timestamp: float
    user_id: str
    value: float

class SlidingWindowAggregator:
    """Sliding window with event-time semantics"""

    def __init__(self, window_size: float, slide_interval: float):
        self.window_size = window_size  # Window size in seconds
        self.slide_interval = slide_interval  # How often to emit
        self.buffers: Dict[str, Deque[Event]] = {}  # Per-key buffers
        self.lock = threading.Lock()

    def process(self, event: Event):
        """Process event and return windows that are ready"""
        with self.lock:
            # Initialize buffer for key
            if event.user_id not in self.buffers:
                self.buffers[event.user_id] = deque()

            buffer = self.buffers[event.user_id]

            
            buffer.append(event)

            
            cutoff = event.timestamp - .window_size
             buffer  buffer[].timestamp < cutoff:
                buffer.popleft()

            
            total = (e.value  e  buffer)
            count = (buffer)
            avg = total / count  count >   

             {
                : event.user_id,
                : cutoff,
                : event.timestamp,
                : total,
                : count,
                : avg
            }


 ():
    aggregator = SlidingWindowAggregator(
        window_size=,     
        slide_interval=    
    )

    
    events = [
        Event(timestamp=, user_id=, value=),
        Event(timestamp=, user_id=, value=),
        Event(timestamp=, user_id=, value=),
        Event(timestamp=, user_id=, value=),  
    ]

     event  events:
        result = aggregator.process(event)
        ()

 __name__ == :
    main()

Pattern 3: Session Windows (Go)

package main

import (
    "fmt"
    "sort"
    "time"
)

type Event struct {
    UserID    string
    Timestamp time.Time
    Value     int
}

type Session struct {
    UserID string
    Start  time.Time
    End    time.Time
    Events []Event
    Sum    int
}

type SessionWindowAggregator struct {
    inactivityGap time.Duration
    sessions      map[string]*Session
}

func NewSessionWindowAggregator(gap time.Duration) *SessionWindowAggregator {
    return &SessionWindowAggregator{
        inactivityGap: gap,
        sessions:      make(map[string]*Session),
    }
}

func (swa *SessionWindowAggregator) Process(event Event) *Session {
    session, exists := swa.sessions[event.UserID]

    if !exists || event.Timestamp.Sub(session.End) > swa.inactivityGap {
        // Start new session
        if exists {
            // Emit completed session
            completed := session
            fmt.Printf("Session complete: %+v\n", completed)
        }

        session = &Session{
            UserID: event.UserID,
            Start:  event.Timestamp,
            End:    event.Timestamp,
            Events: []Event{event},
            Sum:    event.Value,
        }
        swa.sessions[event.UserID] = session
    } else {
        // Extend existing session
        session.End = event.Timestamp
        session.Events = append(session.Events, event)
        session.Sum += event.Value
    }

    return nil
}

 FlushExpired(currentTime time.Time) []*Session {
     expired []*Session

     userID, session :=  swa.sessions {
         currentTime.Sub(session.End) > swa.inactivityGap {
            expired = (expired, session)
            (swa.sessions, userID)
        }
    }

     expired
}

 {
    aggregator := NewSessionWindowAggregator( * time.Minute)

    
    events := []Event{
        {UserID: , Timestamp: time.Now(), Value: },
        {UserID: , Timestamp: time.Now().Add( * time.Minute), Value: },
        {UserID: , Timestamp: time.Now().Add( * time.Minute), Value: }, 
        {UserID: , Timestamp: time.Now().Add( * time.Minute), Value: },
    }

     _, event :=  events {
        aggregator.Process(event)
    }

    
    expired := aggregator.FlushExpired(time.Now().Add( * time.Minute))
     _, session :=  expired {
        fmt.Printf(, session)
    }
}

Pattern 4: Watermark-Based Late Data Handling (Python)

from dataclasses import dataclass
from typing import Dict, List, Optional
import heapq

@dataclass
class Event:
    timestamp: int  # Event time
    key: str
    value: int

@dataclass
class Window:
    start: int
    end: int
    key: str
    sum: int = 0
    count: int = 0

class WatermarkAggregator:
    """Window aggregator with watermark-based triggering"""

    def __init__(
        self,
        window_size: int,
        allowed_lateness: int,
        watermark_delay: int
    ):
        self.window_size = window_size
        self.allowed_lateness = allowed_lateness
        self.watermark_delay = watermark_delay

        self.windows: Dict[tuple, Window] = {}  # (key, window_start) -> Window
        self.max_timestamp = 0
        self.watermark = 0
        self.late_events: List[Event] = []

     () -> [Window]:
        
        
        .max_timestamp = (.max_timestamp, event.timestamp)
        .watermark = .max_timestamp - .watermark_delay

        
        window_start = (event.timestamp // .window_size) * .window_size
        window_end = window_start + .window_size
        window_key = (event.key, window_start)

        
         event.timestamp < .watermark - .allowed_lateness:
            .late_events.append(event)
            ()
             []

        
         window_key   .windows:
            .windows[window_key] = Window(
                start=window_start,
                end=window_end,
                key=event.key
            )

        window = .windows[window_key]
        window. += event.value
        window.count += 

        
         ._emit_completed_windows()

     () -> [Window]:
        
        completed = []
        to_remove = []

         (key, window_start), window  .windows.items():
            
             .watermark > window.end + .allowed_lateness:
                completed.append(window)
                to_remove.append((key, window_start))

        
         key  to_remove:
             .windows[key]

         completed


 ():
    aggregator = WatermarkAggregator(
        window_size=,         
        allowed_lateness=,     
        watermark_delay=       
    )

    
    events = [
        Event(timestamp=, key=, value=),
        Event(timestamp=, key=, value=),
        Event(timestamp=, key=, value=),   
        Event(timestamp=, key=, value=),
        Event(timestamp=, key=, value=),    
    ]

     event  events:
        completed = aggregator.process(event)
         window  completed:
            ()

 __name__ == :
    main()

Pattern 5: Time-Series Aggregation with Downsampling

import numpy as np
from dataclasses import dataclass
from typing import Dict, List
from collections import defaultdict

@dataclass
class TimeSeriesPoint:
    timestamp: int
    metric: str
    value: float

class TimeSeriesAggregator:
    """Multi-resolution time-series aggregation"""

    def __init__(self):
        # Store multiple resolutions
        self.raw: Dict[str, List[TimeSeriesPoint]] = defaultdict(list)
        self.minute: Dict[str, List[tuple]] = defaultdict(list)  # (timestamp, avg, min, max)
        self.hour: Dict[str, List[tuple]] = defaultdict(list)

    def ingest(self, point: TimeSeriesPoint):
        """Ingest raw point and update aggregations"""
        self.raw[point.metric].append(point)

        # Update minute-level aggregation
        minute_bucket = (point.timestamp // 60) * 
        ._update_aggregation(point.metric, minute_bucket, point.value, .minute)

        
        hour_bucket = (point.timestamp // ) * 
        ._update_aggregation(point.metric, hour_bucket, point.value, .hour)

     ():
        
        buckets = storage[metric]

        
          buckets  buckets[-][] != bucket:
            
            buckets.append((bucket, value, value, value, ))  
        :
            
            old_time, old_sum, old_min, old_max, old_count = buckets[-]
            buckets[-] = (
                old_time,
                old_sum + value,
                (old_min, value),
                (old_max, value),
                old_count + 
            )

     () -> []:
        
        storage = {
            : .minute,
            : .hour,
            : .raw
        }[resolution]

         resolution == :
            points = storage[metric]
            filtered = [p  p  points  start <= p.timestamp <= end]
             [(p.timestamp, p.value)  p  filtered]

        buckets = storage[metric]
        filtered = [
            (timestamp, total/count, min_val, max_val)
             timestamp, total, min_val, max_val, count  buckets
             start <= timestamp <= end
        ]

         filtered


 ():
    agg = TimeSeriesAggregator()

    
     i  ():  
        point = TimeSeriesPoint(
            timestamp=i,
            metric=,
            value= +  * np.sin(i / )  
        )
        agg.ingest(point)

    
    minute_data = agg.query(, , , resolution=)
    ()

    hour_data = agg.query(, , , resolution=)
    ()

    
     timestamp, avg, min_val, max_val  hour_data:
        ()

 __name__ == :
    main()

Pattern 6: Top-K Streaming Aggregation

use std::collections::BinaryHeap;
use std::cmp::Reverse;

#[derive(Debug, Clone, Eq, PartialEq)]
struct Item {
    key: String,
    count: usize,
}

impl Ord for Item {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.count.cmp(&other.count)
    }
}

impl PartialOrd for Item {
    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
        Some(self.cmp(other))
    }
}

struct TopKAggregator {
    k: usize,
    counts: std::collections::HashMap<String, usize>,
    heap: BinaryHeap<Reverse<Item>>,
}

impl TopKAggregator {
    fn new(k: usize) -> Self {
        Self {
            k,
            counts: std::collections::HashMap::new(),
            heap: BinaryHeap::new(),
        }
    }

    fn update(& , key: ) {
        
          = .counts.(key.()).();
        *count += ;

        
        .();
    }

     (& ) {
        .heap.();

         (key, count)  &.counts {
              = Item {
                key: key.(),
                count: *count,
            };

             .heap.() < .k {
                .heap.((item));
            }    ((min)) = .heap.() {
                 item.count > min.count {
                    .heap.();
                    .heap.((item));
                }
            }
        }
    }

     (&)  <Item> {
         : <_> = .heap.()
            .(|(item)| item.())
            .();
        items.(|a, b| b.count.(&a.count));
        items
    }
}

 () {
      = TopKAggregator::();

    
      = [, , , , , , ];

       events {
        aggregator.(event.());
    }

    ();
       aggregator.() {
        (, item.key, item.count);
    }
}

Quick Reference

Window Selection Guide

Use Tumbling: Periodic reports, non-overlapping batches
Use Sliding: Moving averages, continuous metrics
Use Session: User behavior, conversation analysis
Use Global: Unbounded state, all-time aggregations

Watermark Strategies

# Bounded delay
watermark = max_timestamp - fixed_delay

# Percentile-based
watermark = percentile(timestamps, 99)  # Allow 1% late

# Heuristic
watermark = max_timestamp - 2 * stddev(inter_arrival_time)

Time Extraction

// Extract event time from data
let event_time = |event: &Event| event.timestamp;

// Assign to window
let window_id = event.timestamp / window_size;

// Session gap check
if current_time - last_event_time > session_gap {
    // Start new session
}

Performance Optimization

Reduce State Size
  → Compact old windows
  → Use approximate algorithms (HyperLogLog, Count-Min Sketch)
  → Expire inactive keys

Batch Processing
  → Buffer events before aggregating
  → Periodic window evaluation

Incremental Updates
  → Use differential dataflow for efficient re-aggregation
  → Maintain summary statistics (sum, count) instead of raw data

Anti-Patterns

❌ NEVER: Use processing time for event-time logic
   → Results depend on processing speed, not actual event timing

❌ NEVER: Wait indefinitely for late data
   → Set allowed lateness bounds

❌ NEVER: Store unbounded state in global windows
   → Use approximate algorithms or periodic cleanup

❌ NEVER: Ignore watermarks
   → Windows never complete, state grows unbounded

❌ NEVER: Use sliding windows with small slide interval on high-volume streams
   → Creates many overlapping windows, high memory usage

❌ NEVER: Recompute entire window on late data
   → Use incremental updates

❌ NEVER: Assume events arrive in order
   → Always design for out-of-order delivery

❌ NEVER: Use session windows without timeouts
   → Sessions never close, memory leak

❌ NEVER: Emit window before watermark passes
   → Incomplete results

❌ NEVER: Drop late data without logging
   → Monitor late data rates for tuning

Related Skills

• timely-dataflow.md - Foundation for windowing with progress tracking
• differential-dataflow.md - Incremental window updates
• dataflow-coordination.md - Watermarks and coordination
• stream-processing.md - High-level stream processing with Kafka

Last Updated: 2025-10-27 Format Version: 1.0 (Atomic)