How-to

Build a streaming topology

A stream is a never-ending flow of events. A topology is the plan: source, ordered stages, sinks. A window groups events by time so you can answer “how many per five minutes?”

Example: count inbound orders every five minutes, by warehouse, and push each bucket to a dashboard feed.

Before you start

Open Streams. Read the readiness banner.

Everything up to windows-on-samples works without live drive. Live consumption of the real river only happens when an administrator enables it.

1. Define the topology

Load starter, then adapt:

{
  "id": "orders-per-5m",
  "name": "Orders per 5 minutes",
  "source_id": "orders-kafka",
  "stages": [
    { "id": "keep-creates", "component": "filter-creates" },
    { "id": "by-warehouse", "component": "extract-warehouse" }
  ],
  "sinks": [
    { "id": "dashboard-feed", "component": "orders-dashboard-sink" }
  ]
}

source_id is a streaming source you already configured. Each component is a small Sirius component. Validate, then Save. The topology is disabled — a safe draft.

2. Plan

Plan resolves source → stages → sinks and warns about dangling ids. Fix names before you run.

3. Sample-run

Paste one record. Run. You see what each stage produced. Sinks are skipped — nothing is emitted downstream.

{ "event_type": "order.created", "warehouse": "west", "id": "A-1" }

Try a record that should be filtered out and confirm it dies in the first stage.

4. Add a window

{
  "id": "count-5m",
  "window": { "type": "tumbling", "size": "5m" },
  "aggregate": "count",
  "group_by": "warehouse"
}

• Tumbling — back-to-back buckets (2:00–2:05, 2:05–2:10).
• Sliding — overlapping. Also driven live.
• Session — gap-based. Design-time aggregation only; the live engine does not drive sessions.

Save, then aggregate a batch of samples in the Window panel so you see the shape of the output.

5. Go live (when live drive is on)

1. Confirm the banner’s live-drive line.
2. Confirm the source is consuming and the topology has a windowed stage. The live engine drives windows; a topology of only plain transforms will not consume the river.
3. Enable. Watch State: offset, watermark, open windows.
4. When a window closes, counts go to the sink exactly once, including across restarts.

Late records arriving after the watermark (plus any configured lateness grace) are not counted. Set grace to the lateness you actually observe.

# Dry-run (default). Add "save": true in the body to persist.
curl -s -X POST "$SIRIUS/streams/topologies" \
  -H 'content-type: application/json' --data-binary @topology.json

curl -s "$SIRIUS/streams/topologies/orders-per-5m/plan"
curl -s -X POST "$SIRIUS/streams/topologies/orders-per-5m/run" \
  -H 'content-type: application/json' \
  -d '{"record":{"event_type":"order.created","warehouse":"west"}}'
curl -s -X POST "$SIRIUS/streams/topologies/orders-per-5m/enable"
curl -s "$SIRIUS/streams/topologies/orders-per-5m/state"

If a route exists in the catalog but is not wired here, you get 501 not_configured. That is configuration, not a missing product.