Aller au contenu principal

Fenêtres & agrégations

Intermédiaire ⏱ 15 min streaming · fenêtres · agrégations

Vous allez construire : un stage streaming qui compte et profile les événements par clé dans une fenêtre glissante — la primitive derrière la vélocité de fraude, la santé de flotte et les dashboards temps réel.

webhook → streaming:window5 → (agrégats fenêtrés en sortie)

Prérequis​

1 · Générez avec un stage streaming​

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

2 · Déclarez la fenêtre​

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)"

Deux opérateurs : keyBy donne à chaque user_id sa propre fenêtre indépendante, puis window déclare la forme. Les specs de fenêtre : tumbling(60s) · sliding(60s,10s) · session(30s) · count(100) · count_sliding(100,10) · global(). La palette d'agrégateurs : count() · sum · min · max · avg · first · last · stddev · countWhere(expr) · sumWhere(expr). La spec et chaque agrégation sont validées au deploy — une faute de frappe est une erreur nommée, pas un no-op silencieux.

3 · Déployez et envoyez une rafale​

pulse secret set PULSE_WEBHOOK_SECRET dev-secret   # la source webhook générée l'exige
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 · Observez les agrégats​

pulse events tail --topic velocity.window5.out

Les événements d'agrégat pour la clé u1 : events grimpe de 1 à 5, total s'accumule, big_ones compte les envois >100, spread (stddev) s'élargit à mesure que les montants divergent.

5 · Prouvez l'isolation​

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

u2 démarre sa propre fenêtre à 1 événement — les chiffres de u1 sont intacts. L'état par clé est par clé, toujours.

Ce qui vient de se passer​

Vous avez déclaré du traitement de flux avec état sans écrire de code d'état. Le moteur maintient des fenêtres par clé, applique la sémantique event-time avec des watermarks (les événements en retard ne corrompent pas silencieusement la fenêtre — les vrais retardataires partent en dead-letter queue), et émet les agrégats en aval comme n'importe quel événement. Chaînez un stage rule-based derrière et vous avez un détecteur.

Dépannage​

Aucun agrégat en sortie

Vérifiez que le nom du stage dans votre commande tail correspond au YAML (velocity.window5.out). Les agrégats s'émettent au fil des mises à jour de la fenêtre — envoyez plus d'un événement.

Le deploy rejette ma spec de fenêtre

La spec fenêtre/agrégation est validée au deploy — l'erreur nomme le champ. Vérifiez l'orthographe de l'agrégateur contre la palette de l'étape 2.

Ensuite​