🌊 Stream Operators - Fluent API Guide

A powerful, composable API for building stream processing pipelines in Rust. Inspired by Apache Flink, Kafka Streams, and functional programming patterns.

Overview
Core Concepts
Basic Operators
Advanced Operators
Windowing
Aggregations
Real-World Examples
Performance Tips

Overview

Stream Operators provide a fluent, chainable API for processing event streams with functional-style transformations, aggregations, and windowing operations.

Key Features

✅ Fluent API: Chain operations naturally like Rust iterators
✅ Type-Safe: Leverages Rust's type system for compile-time safety
✅ Zero-Copy: Efficient event processing with minimal allocations
✅ Composable: Build complex pipelines from simple operators
✅ Functional: Map, filter, reduce, and more
✅ Windowing: Time-based aggregations (sliding, tumbling, session)
✅ Key-By: Partition streams by key for parallel processing

Core Concepts

DataStream

The fundamental abstraction for a stream of events:

use rust_rule_engine::streaming::*;

// Create from events
let stream = DataStream::from_events(events);

// Create empty and build up
let mut stream = DataStream::new();
stream.push(event);

KeyedStream

A stream partitioned by a key field:

let keyed = stream.key_by(|e| e.get_string("user_id").unwrap_or("").to_string());

WindowedStream

A stream with time-based windowing applied:

let windowed = stream.window(WindowConfig::tumbling(Duration::from_secs(60)));

Basic Operators

Filter

Keep only events matching a predicate:

let filtered = stream
    .filter(|e| e.get_numeric("amount").unwrap_or(0.0) > 100.0);

Use Cases:

Remove invalid data
Focus on specific event types
Apply business rules

Map

Transform each event:

let transformed = stream
    .map(|mut e| {
        // Add computed field
        if let Some(price) = e.get_numeric("price") {
            let tax = price * 0.1;
            e.data.insert("tax".to_string(), Value::Number(tax));
        }
        e
    });

Use Cases:

Enrich events
Format data
Add derived fields

FlatMap

Transform each event into multiple events:

let expanded = stream
    .flat_map(|e| {
        // Split one event into multiple
        let mut events = Vec::new();
        if let Some(items) = e.get_array("items") {
            for item in items {
                let mut new_event = e.clone();
                new_event.data.insert("item".to_string(), item.clone());
                events.push(new_event);
            }
        }
        events
    });

Use Cases:

Unnesting arrays
Event explosion
Normalization

ForEach

Execute side effects without modifying the stream:

let stream = stream
    .for_each(|e| {
        println!("Processing: {:?}", e);
        // Log, send metrics, etc.
    });

Use Cases:

Logging
Metrics collection
Debugging

Advanced Operators

Key-By

Partition stream by a key for grouped operations:

let keyed = stream
    .key_by(|e| e.get_string("user_id").unwrap_or("unknown").to_string());

// Then aggregate per key
let totals = keyed.aggregate(Sum::new("amount"));

Use Cases:

Per-user analytics
Session tracking
Grouped aggregations

Reduce

Combine events into a single result:

let total = stream
    .reduce(|mut acc, e| {
        let acc_val = acc.get_numeric("total").unwrap_or(0.0);
        let e_val = e.get_numeric("amount").unwrap_or(0.0);
        acc.data.insert("total".to_string(), Value::Number(acc_val + e_val));
        acc
    });

Use Cases:

Running totals
Accumulation
Finding max/min

Group-By

Group events by a key and apply operations:

let grouped = stream
    .group_by(|e| e.get_string("category").unwrap_or("").to_string());

// Count per group
let counts = grouped.count();

// Aggregate per group
let averages = grouped.aggregate(Average::new("price"));

Use Cases:

Category analysis
Cohort grouping
Distribution analysis

Union

Combine two streams:

let combined = stream1.union(stream2);

Use Cases:

Merge multiple sources
Combine filtered results
Data consolidation

Take / Skip

Limit stream size:

let first_10 = stream.take(10);
let skip_first_5 = stream.skip(5);

Use Cases:

Sampling
Pagination
Testing

Sort

Order events by a key:

let sorted = stream
    .sort_by(|e| e.metadata.timestamp);

Use Cases:

Time ordering
Priority sorting
Ranking

Windowing

Apply time-based windows for aggregations over time ranges.

Tumbling Windows

Non-overlapping fixed-size windows:

let windowed = stream
    .window(WindowConfig::tumbling(Duration::from_secs(60)));

// Aggregate within each window
let results = windowed.aggregate(Sum::new("amount"));

Use Cases:

Hourly/daily summaries
Batch processing
Report generation

Example:

Events: |--A--B--|--C--D--|--E--F--|
Window:    [W1]     [W2]     [W3]

Sliding Windows

Overlapping windows that slide forward:

let windowed = stream
    .window(WindowConfig::sliding(Duration::from_secs(60)));

Use Cases:

Moving averages
Trend detection
Smoothing

Example:

Events: |--A--B--C--D--E--F--|
Window:    [----W1----]
              [----W2----]
                 [----W3----]

Session Windows

Windows based on inactivity gaps:

let windowed = stream
    .window(WindowConfig::session(Duration::from_secs(300)));

Use Cases:

User sessions
Activity bursts
Click streams

Window Configuration

Customize window behavior:

let config = WindowConfig::tumbling(Duration::from_secs(60))
    .with_max_events(10000);  // Limit events per window

Window Operations

// Count per window
let counts = windowed.counts();

// Aggregate per window
let sums = windowed.aggregate(Sum::new("amount"));

// Reduce per window
let results = windowed.reduce(|acc, e| /* ... */);

// Flatten back to stream
let flattened = windowed.flatten();

Aggregations

Built-in aggregation functions for stream analysis.

Count

Count events:

let count = stream.count();

// Or as aggregation
let result = stream.aggregate(Count);

Sum

Sum numeric values:

let total = stream.aggregate(Sum::new("amount"));

Average

Calculate average:

let avg = stream.aggregate(Average::new("price"));

Min / Max

Find minimum or maximum:

let min_price = stream.aggregate(Min::new("price"));
let max_price = stream.aggregate(Max::new("price"));

Custom Aggregator

Create custom aggregation logic:

let custom = CustomAggregator::new(|events: &[StreamEvent]| {
    // Your custom logic
    let values: Vec<f64> = events
        .iter()
        .filter_map(|e| e.get_numeric("value"))
        .collect();
    
    let sum: f64 = values.iter().sum();
    let avg = sum / values.len() as f64;
    
    let mut result = HashMap::new();
    result.insert("sum".to_string(), Value::Number(sum));
    result.insert("avg".to_string(), Value::Number(avg));
    
    AggregateResult::Map(result)
});

let result = stream.aggregate(custom);

Real-World Examples

Example 1: E-Commerce Analytics

Calculate revenue per category with discounts:

let revenue_by_category = DataStream::from_events(transactions)
    .filter(|e| e.get_string("status") == Some("completed"))
    .map(|mut e| {
        // Apply discount
        if let Some(amount) = e.get_numeric("amount") {
            let discount = e.get_numeric("discount").unwrap_or(0.0);
            let final_amount = amount * (1.0 - discount);
            e.data.insert("final_amount".to_string(), Value::Number(final_amount));
        }
        e
    })
    .key_by(|e| e.get_string("category").unwrap_or("").to_string())
    .aggregate(Sum::new("final_amount"));

for (category, total) in revenue_by_category {
    if let Some(amount) = total.as_number() {
        println!("Category {}: ${:.2}", category, amount);
    }
}

Example 2: IoT Sensor Monitoring

Detect temperature anomalies in real-time:

let alerts = DataStream::from_events(sensor_readings)
    .window(WindowConfig::tumbling(Duration::from_secs(60)))
    .aggregate(CustomAggregator::new(|events| {
        let temps: Vec<f64> = events
            .iter()
            .filter_map(|e| e.get_numeric("temperature"))
            .collect();
        
        let avg = temps.iter().sum::<f64>() / temps.len() as f64;
        let max = temps.iter().fold(f64::NEG_INFINITY, |a, &b| a.max(b));
        
        let anomaly = max > avg + 20.0;  // 20°C spike
        
        let mut result = HashMap::new();
        result.insert("has_anomaly".to_string(), Value::Boolean(anomaly));
        result.insert("max_temp".to_string(), Value::Number(max));
        
        AggregateResult::Map(result)
    }));

Example 3: User Session Analysis

Track user behavior patterns:

let session_stats = DataStream::from_events(clickstream)
    .filter(|e| e.event_type == "PageView")
    .key_by(|e| e.get_string("user_id").unwrap_or("").to_string())
    .window(WindowConfig::session(Duration::from_secs(1800)))  // 30 min timeout
    .aggregate(CustomAggregator::new(|events| {
        let pages_viewed = events.len();
        let duration = if !events.is_empty() {
            let first = events.first().unwrap().metadata.timestamp;
            let last = events.last().unwrap().metadata.timestamp;
            last - first
        } else {
            0
        };
        
        let mut result = HashMap::new();
        result.insert("pages".to_string(), Value::Number(pages_viewed as f64));
        result.insert("duration_ms".to_string(), Value::Number(duration as f64));
        
        AggregateResult::Map(result)
    }));

Example 4: Fraud Detection

Detect suspicious transaction patterns:

let suspicious = DataStream::from_events(transactions)
    .key_by(|e| e.get_string("user_id").unwrap_or("").to_string())
    .window(WindowConfig::sliding(Duration::from_secs(300)))  // 5 min window
    .aggregate(CustomAggregator::new(|events| {
        let count = events.len();
        let total: f64 = events
            .iter()
            .filter_map(|e| e.get_numeric("amount"))
            .sum();
        
        // Flag if: >10 transactions OR total > $5000 in 5 minutes
        let is_suspicious = count > 10 || total > 5000.0;
        
        let mut result = HashMap::new();
        result.insert("suspicious".to_string(), Value::Boolean(is_suspicious));
        result.insert("tx_count".to_string(), Value::Number(count as f64));
        result.insert("total".to_string(), Value::Number(total));
        
        AggregateResult::Map(result)
    }));

Performance Tips

1. Use Filter Early

Filter unwanted events as early as possible in the pipeline:

// ✅ Good - filter first
stream
    .filter(|e| e.event_type == "Purchase")
    .map(|e| expensive_transformation(e))
    .aggregate(Sum::new("amount"));

// ❌ Bad - transform everything
stream
    .map(|e| expensive_transformation(e))
    .filter(|e| e.event_type == "Purchase")
    .aggregate(Sum::new("amount"));

2. Avoid Unnecessary Clones

Use references when possible:

// ✅ Good - no clone in filter
stream.filter(|e| e.get_numeric("amount").unwrap_or(0.0) > 100.0)

// ❌ Bad - unnecessary clone
stream.filter(|e| {
    let cloned = e.clone();  // Unnecessary!
    cloned.get_numeric("amount").unwrap_or(0.0) > 100.0
})

3. Batch Operations

Process events in batches when possible:

// Use windowing to batch events
stream
    .window(WindowConfig::tumbling(Duration::from_secs(1)))
    .aggregate(custom_batch_aggregator)

4. Limit Data Size

Use take() for testing or sampling:

stream
    .take(1000)  // Process only first 1000
    .aggregate(Sum::new("amount"))

5. Choose Appropriate Window Types

Tumbling: Best for non-overlapping summaries (lowest memory)
Sliding: For moving averages (higher memory, overlapping results)
Session: For user behavior (variable window size)

6. Key-By Cardinality

Be mindful of key cardinality when using key_by():

// ✅ Good - low cardinality (user_ids)
stream.key_by(|e| e.get_string("user_id").unwrap_or("").to_string())

// ⚠️  Caution - high cardinality (unique event IDs)
stream.key_by(|e| e.id.clone())  // Creates too many groups!

API Reference

DataStream Methods

Method	Description	Returns
`filter(predicate)`	Keep events matching predicate	`DataStream`
`map(mapper)`	Transform each event	`DataStream`
`flat_map(mapper)`	Transform to multiple events	`DataStream`
`key_by(selector)`	Partition by key	`KeyedStream<K>`
`window(config)`	Apply time window	`WindowedStream`
`reduce(reducer)`	Combine to single result	`Option<StreamEvent>`
`group_by(selector)`	Group events	`GroupedStream<K>`
`aggregate(aggregator)`	Apply aggregation	`AggregateResult`
`for_each(action)`	Execute side effect	`DataStream`
`union(other)`	Combine streams	`DataStream`
`take(n)`	Take first n events	`DataStream`
`skip(n)`	Skip first n events	`DataStream`
`sort_by(key_fn)`	Sort events	`DataStream`
`count()`	Count events	`usize`
`collect()`	Collect to Vec	`Vec<StreamEvent>`

KeyedStream Methods

Method	Description	Returns
`reduce(reducer)`	Reduce per key	`HashMap<K, StreamEvent>`
`aggregate(aggregator)`	Aggregate per key	`HashMap<K, AggregateResult>`
`window(config)`	Window per key	`KeyedWindowedStream<K>`
`count()`	Count per key	`HashMap<K, usize>`
`flatten()`	Back to DataStream	`DataStream`

WindowedStream Methods

Method	Description	Returns
`aggregate(aggregator)`	Aggregate per window	`Vec<AggregateResult>`
`reduce(reducer)`	Reduce per window	`Vec<StreamEvent>`
`counts()`	Count per window	`Vec<usize>`
`flatten()`	Back to DataStream	`DataStream`

Examples

See full examples:

examples/03-advanced-features/stream_operators_demo.rs - Basic operators
examples/06-use-cases/iot_monitoring_demo.rs - IoT monitoring
examples/06-use-cases/fraud_detection_stream.rs - Fraud detection

Next Steps

Explore State Management for stateful operators
Learn about Watermarks for handling late data
Check out Complex Event Processing for pattern detection

FilesExpand file tree

STREAM_OPERATORS.md

Latest commit

History

STREAM_OPERATORS.md

File metadata and controls

🌊 Stream Operators - Fluent API Guide

Table of Contents

Overview

Key Features

Core Concepts

DataStream

KeyedStream

WindowedStream

Basic Operators

Filter

Map

FlatMap

ForEach

Advanced Operators

Key-By

Reduce

Group-By

Union

Take / Skip

Sort

Windowing

Tumbling Windows

Sliding Windows

Session Windows

Window Configuration

Window Operations

Aggregations

Count

Sum

Average

Min / Max

Custom Aggregator

Real-World Examples

Example 1: E-Commerce Analytics

Example 2: IoT Sensor Monitoring

Example 3: User Session Analysis

Example 4: Fraud Detection

Performance Tips

1. Use Filter Early

2. Avoid Unnecessary Clones

3. Batch Operations

4. Limit Data Size

5. Choose Appropriate Window Types

6. Key-By Cardinality

API Reference

DataStream Methods

KeyedStream Methods

WindowedStream Methods

Examples

Next Steps