Skip to main content

Windows & aggregations

Intermediate ⏱ 15 min streaming · windows · aggregations

You will build: a streaming stage that counts and profiles events per key in a sliding window — the primitive behind fraud velocity, fleet health and live dashboards.

webhook → streaming:window5 → (windowed aggregates out)

Prerequisites​

1 · Scaffold with a streaming stage​

pulse new velocity --source webhook --stage streaming:window5

2 · Declare the window​

stages:
  - name: window5
    engine: streaming
    operators:
      - { type: keyBy, field: user_id }
      - type: window
        spec: "sliding(5m,1m)"
        aggregations:
          events:    "count()"
          total:     "sum(amount)"
          big_ones:  "countWhere(amount > 100)"
          spread:    "stddev(amount)"

Two operators: keyBy gives every user_id its own independent window, then window declares the shape. Window specs: tumbling(60s) · sliding(60s,10s) · session(30s) · count(100) · count_sliding(100,10) · global(). The aggregator palette: count() · sum · min · max · avg · first · last · stddev · countWhere(expr) · sumWhere(expr). Both the spec and every aggregation are validated at deploy time — a typo is a named error, not a silent no-op.

3 · Deploy and send a burst​

pulse secret set PULSE_WEBHOOK_SECRET dev-secret   # the scaffolded webhook source requires it
pulse deploy .
for i in 1 2 3 4 5; do
  pulse events publish --topic velocity.in --value '{"user_id":"u1","amount":'"$((i*40))"'}'
done

4 · Watch the aggregates​

pulse events tail --topic velocity.window5.out

Aggregate events for key u1: events climbing 1→5, total accumulating, big_ones counting the >100 sends, spread (stddev) widening as amounts diverge.

5 · Prove the isolation​

pulse events publish --topic velocity.in --value '{"user_id":"u2","amount":10}'

u2 starts its own window at 1 event — u1's numbers are untouched. Keyed state is per key, always.

What just happened​

You declared stateful stream processing without writing state code. The engine maintains per-key windows, applies event-time semantics with watermarks (late events don't silently corrupt the window — genuinely late ones route to a dead-letter queue), and emits aggregate events downstream like any other event. Chain a rule-based stage after this and you have a detector.

Troubleshooting​

No aggregate output appears

Check the stage name in your tail command matches the YAML (velocity.window5.out). Aggregates emit as the window updates — send more than one event.

Deploy rejects my window spec

The window/aggregation spec is validated at deploy time — the error names the field. Check the aggregator spelling against the palette in step 2.

Next​