Aggregation

Prerequisites: Batching and Windowing, Key Concepts

Aggregation groups items by a key and a time window, accumulates state per group, and emits results when windows close.

How Aggregation Works

1. Each item is assigned a key (via GetKey) and a window (via the configured WindowAssigner)
2. Per (key, window) pair, an accumulator is maintained
3. Each item updates its accumulator via Accumulate
4. When a watermark closes a window, GetResult is called and the result is emitted downstream

Watermarks use the same event time as window assignment: the item's ITimestamped.Timestamp, or the configured TimestampExtractor, or arrival time when neither is available. Historical, replayed or back-filled data therefore aggregates correctly, because window closing follows the data's own clock and not the wall clock.

Items arriving after their window closes are dropped to prevent multiple partial results. The number of dropped items is reported by the node's LateItemsDropped property.

Writing an Aggregate Node

Extend AggregateNode<TIn, TKey, TResult> when the accumulator type is the same as the result:

public class SalesTotal : AggregateNode<Sale, string, SalesSummary>
{
    public SalesTotal() : base(new AggregateNodeConfiguration<Sale>(
        WindowAssigner.Tumbling(TimeSpan.FromHours(1)),
        TimestampExtractor: sale => sale.Timestamp)) { }

    public override string GetKey(Sale item) => item.Category;

    public override SalesSummary CreateAccumulator()
        => new(Count: 0, Total: 0m);

    public override SalesSummary Accumulate(SalesSummary acc, Sale item)
        => acc with { Count = acc.Count + 1, Total = acc.Total + item.Amount };
}

Register and connect:

var source = builder.AddSource<SaleSource, Sale>("sales");
var aggregate = builder.AddAggregate<SalesTotal, Sale, string, SalesSummary>("totals");
var sink = builder.AddSink<ReportSink, SalesSummary>("report");

builder.Connect(source, aggregate);
builder.Connect(aggregate, sink);

Advanced Aggregation

When the accumulator differs from the result (e.g., you accumulate a list but emit a computed summary), extend AdvancedAggregateNode<TIn, TKey, TAccumulate, TResult>:

public class StatsAggregator
    : AdvancedAggregateNode<Measurement, string, RunningStats, StatsSummary>
{
    public StatsAggregator() : base(new AggregateNodeConfiguration<Measurement>(
        WindowAssigner.Sliding(TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(1)),
        TimestampExtractor: m => m.Timestamp)) { }

    public override string GetKey(Measurement item) => item.SensorId;
    public override RunningStats CreateAccumulator() => new();
    public override RunningStats Accumulate(RunningStats stats, Measurement item)
        => stats.Add(item.Value);
    public override StatsSummary GetResult(RunningStats stats)
        => new(stats.Mean, stats.StdDev, stats.Count);
}

Lambda-Based Aggregation

For simple cases, use the fluent grouping API:

// Tumbling window
var totals = builder.GroupItems<Sale>()
    .ForTemporalCorrectness(
        windowSize: TimeSpan.FromHours(1),
        keySelector: sale => sale.Category,
        initialValue: () => 0m,
        accumulator: (sum, sale) => sum + sale.Amount,
        timestampExtractor: sale => sale.Timestamp);

// Sliding window
var averages = builder.GroupItems<Measurement>()
    .ForTemporalCorrectnessWithSlidingWindows(
        windowSize: TimeSpan.FromMinutes(5),
        slide: TimeSpan.FromMinutes(1),
        keySelector: m => m.SensorId,
        initialValue: () => (Sum: 0.0, Count: 0),
        accumulator: (acc, m) => (acc.Sum + m.Value, acc.Count + 1),
        timestampExtractor: m => m.Timestamp);

Configuration Options

AggregateNodeConfiguration<TIn> controls window behavior:

Property	Default	Description
WindowAssigner	(required)	Tumbling or sliding window strategy. Window sizes and slides must be positive
TimestampExtractor	null	Extracts event time from items; uses arrival time for items that are neither ITimestamped nor extracted
MaxOutOfOrderness	5 minutes	How late an item can arrive and still be included. Must not be negative

The watermark is re-evaluated on every item, so windows close as soon as the data passes them, even in a fast replay.

Windows are aligned on UTC ticks, so the same instant always lands in the same window regardless of the timestamp's UTC offset.

Monitoring Aggregation

Call GetMetrics() on your aggregate node to inspect state:

var (totalProcessed, totalClosed, maxConcurrent) = aggregateNode.GetMetrics();
int activeWindows = aggregateNode.GetActiveWindowCount();
long lateItems = aggregateNode.LateItemsDropped;

Next Steps

• Batching and Windowing - window types and watermark mechanics
• Streaming Large Datasets - memory management for windowed aggregation