Stream processing handles data in motion — events arriving continuously rather than sitting in a file waiting to be read. The challenge is computing meaningful results (counts, averages, anomalies) over infinite streams without storing everything in memory.
The key concepts are: windowing (grouping events by time for aggregation), state (what you remember across events), and time (the difference between when events were produced vs when you process them). Go’s channel primitives map naturally onto stream processing patterns.
For file-based batch processing see Go working with large datasets. For worker pool patterns see Go worker pools.
The Basic Stream Pipeline
A stream processor is a pipeline: events flow in, transformations are applied, results flow out. Each stage in the pipeline runs in its own goroutine, communicating via channels. When a stage is slow, the channel buffers absorb bursts; when the buffer fills, backpressure slows the upstream stage.
type Event struct {
Timestamp time.Time
Key string
Value float64
}
// A pipeline stage: reads from in, writes to out
type Stage func(ctx context.Context, in <-chan Event) <-chan Event
// Chain stages together
func pipeline(ctx context.Context, source <-chan Event, stages ...Stage) <-chan Event {
current := source
for _, stage := range stages {
current = stage(ctx, current)
}
return current
}
This pattern composes stages cleanly. Each stage is independent and testable in isolation — you can test it with a synthetic input channel and verify the output channel.
Tumbling Windows: Non-Overlapping Time Buckets
A tumbling window groups all events in a fixed time interval into one batch. Events in window N never appear in window N+1. Common uses: compute per-minute request rates, hourly revenue totals, daily active users.
// TumblingWindow collects events into fixed-duration buckets and emits each bucket
func TumblingWindow(ctx context.Context, in <-chan Event, size time.Duration) <-chan []Event {
out := make(chan []Event)
go func() {
defer close(out)
var window []Event
ticker := time.NewTicker(size)
defer ticker.Stop()
for {
select {
case event, ok := <-in:
if !ok {
// Input closed — emit remaining events
if len(window) > 0 {
out <- window
}
return
}
window = append(window, event)
case <-ticker.C:
// Window expired — emit and start fresh
if len(window) > 0 {
out <- window
window = nil
}
case <-ctx.Done():
return
}
}
}()
return out
}
// Usage: compute average value per minute
for bucket := range TumblingWindow(ctx, events, time.Minute) {
var sum float64
for _, e := range bucket {
sum += e.Value
}
fmt.Printf("minute avg: %.2f (%d events)\n", sum/float64(len(bucket)), len(bucket))
}
The ticker drives window boundaries. Events received after the ticker fires but before the window is re-initialized belong to the new window. This is processing-time windowing — boundaries are based on when events arrive, not when they were produced.
Sliding Windows: Overlapping Time Ranges
A sliding window emits a snapshot of all events in the last N time units, updated every M time units. Unlike tumbling windows (where each event belongs to exactly one window), sliding windows allow events to appear in multiple windows.
Use case: “requests in the last 5 minutes” updated every 30 seconds for a dashboard metric.
// SlidingWindow maintains a rolling buffer of events within the window duration
// and emits a snapshot every slideInterval
func SlidingWindow(ctx context.Context, in <-chan Event, windowDur, slideInterval time.Duration) <-chan []Event {
out := make(chan []Event)
go func() {
defer close(out)
var buffer []Event
slide := time.NewTicker(slideInterval)
defer slide.Stop()
evict := func() {
cutoff := time.Now().Add(-windowDur)
i := 0
for i < len(buffer) && buffer[i].Timestamp.Before(cutoff) {
i++
}
buffer = buffer[i:]
}
for {
select {
case event, ok := <-in:
if !ok {
return
}
buffer = append(buffer, event)
case <-slide.C:
evict() // remove events older than window duration
if len(buffer) > 0 {
snapshot := make([]Event, len(buffer))
copy(snapshot, buffer)
out <- snapshot
}
case <-ctx.Done():
return
}
}
}()
return out
}
The evict function removes events that have aged out of the window. Events use their Timestamp field (production time) for eviction — this is event-time windowing. If events arrive out of order (late data), they’re still placed correctly in the window until the window boundary advances past them.
Session Windows: Activity-Based Grouping
Session windows group events by activity rather than fixed time. A session ends when there’s no new activity for a configurable timeout. This is ideal for user session tracking — group all page views within 30 minutes of each other as one session:
// SessionWindow groups events into sessions separated by inactivity gaps
func SessionWindow(ctx context.Context, in <-chan Event, timeout time.Duration) <-chan []Event {
out := make(chan []Event)
go func() {
defer close(out)
var session []Event
timer := time.NewTimer(timeout)
defer timer.Stop()
timerActive := false
for {
select {
case event, ok := <-in:
if !ok {
if len(session) > 0 {
out <- session
}
return
}
if !timerActive {
timerActive = true
} else {
// Reset timer on each new event
if !timer.Stop() {
select {
case <-timer.C:
default:
}
}
timer.Reset(timeout)
}
session = append(session, event)
case <-timer.C:
// Inactivity timeout — session complete
if len(session) > 0 {
out <- session
session = nil
timerActive = false
}
case <-ctx.Done():
if len(session) > 0 {
out <- session
}
return
}
}
}()
return out
}
Each event resets the inactivity timer. When the timer fires without a new event, the session ends and is emitted downstream. Sessions can be arbitrarily long (many events over a long period) or short (single event).
Stateful Aggregation
Some computations require remembering state across events — running totals, counts per key, deduplication. State must be protected if multiple goroutines access it:
type KeyedAggregator struct {
mu sync.RWMutex
state map[string]*KeyState
}
type KeyState struct {
Count int
Sum float64
Min float64
Max float64
}
func NewKeyedAggregator() *KeyedAggregator {
return &KeyedAggregator{state: make(map[string]*KeyState)}
}
func (a *KeyedAggregator) Process(event Event) {
a.mu.Lock()
defer a.mu.Unlock()
s, ok := a.state[event.Key]
if !ok {
s = &KeyState{Min: event.Value, Max: event.Value}
a.state[event.Key] = s
}
s.Count++
s.Sum += event.Value
if event.Value < s.Min { s.Min = event.Value }
if event.Value > s.Max { s.Max = event.Value }
}
func (a *KeyedAggregator) Snapshot(key string) (KeyState, bool) {
a.mu.RLock()
defer a.mu.RUnlock()
s, ok := a.state[key]
if !ok {
return KeyState{}, false
}
return *s, true
}
For large state that must survive process restarts, persist to Redis or a database. For ephemeral in-process state, this map-with-mutex pattern is sufficient.
State also accumulates without bound if keys are never retired. Add a cleanup routine that removes keys inactive for longer than your retention window:
func (a *KeyedAggregator) Evict(olderThan time.Time) {
// Track last-seen time per key and remove stale entries
}
Integrating with Kafka
For production event streams, data comes from a message broker. The confluent-kafka-go or segmentio/kafka-go libraries provide Kafka consumers that feed into your stream processing pipeline:
import "github.com/segmentio/kafka-go"
func consumeKafka(ctx context.Context, brokers []string, topic string) <-chan Event {
out := make(chan Event, 1000)
go func() {
defer close(out)
r := kafka.NewReader(kafka.ReaderConfig{
Brokers: brokers,
Topic: topic,
GroupID: "stream-processor-v1",
MinBytes: 10e3, // 10 KB — wait for some data to accumulate
MaxBytes: 10e6, // 10 MB max per batch
CommitInterval: time.Second,
})
defer r.Close()
for {
msg, err := r.FetchMessage(ctx)
if err != nil {
if ctx.Err() != nil {
return // context cancelled — clean shutdown
}
slog.Error("kafka fetch", slog.Any("error", err))
continue
}
var event Event
if err := json.Unmarshal(msg.Value, &event); err != nil {
slog.Warn("skip bad message", slog.Any("error", err))
r.CommitMessages(ctx, msg)
continue
}
select {
case out <- event:
r.CommitMessages(ctx, msg)
case <-ctx.Done():
return
}
}
}()
return out
}
// Wire everything together
func main() {
ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGTERM)
defer cancel()
events := consumeKafka(ctx, []string{"kafka:9092"}, "sensor-readings")
// 1-minute tumbling windows
windows := TumblingWindow(ctx, events, time.Minute)
// Aggregate each window
for bucket := range windows {
agg := aggregate(bucket)
slog.Info("window complete",
slog.Int("events", len(bucket)),
slog.Float64("avg", agg.Avg),
slog.Float64("max", agg.Max),
)
}
}
Handling Late Data
In real systems, events arrive out of order — network delays, clock skew, retries. Event-time processing must decide how long to wait for late data before closing a window.
The standard approach is a watermark — a time threshold below which late events are either dropped or sent to a separate “late bucket”:
type WatermarkedWindow struct {
windowSize time.Duration
lateTolerance time.Duration
windows map[time.Time][]Event // window-start → events
watermark time.Time
}
func (w *WatermarkedWindow) Add(event Event) (completed []Event, isLate bool) {
windowStart := event.Timestamp.Truncate(w.windowSize)
// Event falls before watermark — it's late
if windowStart.Before(w.watermark) {
return nil, true
}
w.windows[windowStart] = append(w.windows[windowStart], event)
// Advance watermark and emit completed windows
candidateWatermark := event.Timestamp.Add(-w.lateTolerance)
if candidateWatermark.After(w.watermark) {
w.watermark = candidateWatermark
// Emit and remove windows that are now behind the watermark
for start, events := range w.windows {
if start.Add(w.windowSize).Before(w.watermark) {
completed = append(completed, events...)
delete(w.windows, start)
}
}
}
return completed, false
}
The lateTolerance sets how long after the window’s nominal end time you wait before declaring it closed. A larger tolerance catches more late events but increases latency.
Summary
- Tumbling windows group events into fixed, non-overlapping time buckets — emit when the ticker fires
- Sliding windows maintain a rolling buffer and emit snapshots on a slide interval — each event can appear in multiple windows
- Session windows close on inactivity — each event resets the timer; useful for user behavior analysis
- Use event timestamps (not arrival time) for windowing when events can arrive out of order
- Protect shared aggregation state with a mutex; add eviction to prevent unbounded growth
- Wire to Kafka or another broker for production: the consumer produces a
<-chan Eventthat feeds directly into window functions
Resources
- Go Blog: Pipelines and cancellation
- segmentio/kafka-go
- Apache Beam windowing concepts
- Streaming Systems (book) — Tyler Akidau et al.
Comments