Stream Processor

High-performance, production-ready stream processing library for LLM Auto-Optimizer.

Features

• Windowing: Tumbling, Sliding, and Session windows
• Aggregations: Count, Sum, Average, Min, Max, Percentiles, Standard Deviation
• Time Semantics: Event-time processing with watermark support
• Late Events: Configurable handling for out-of-order events
• Performance: Sub-100ms latency, 100K+ events/sec throughput
• Statistics: Online variance, EMA, sliding window stats, anomaly detection

Quick Start

use processor::{StreamProcessorBuilder, aggregation::CompositeAggregator};
use chrono::Duration;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Create a stream processor with 5-minute tumbling windows
    let mut processor = StreamProcessorBuilder::new()
        .build_tumbling(
            Duration::minutes(5),
            |_key| {
                let mut agg = CompositeAggregator::new();
                agg.add_count("requests")
                   .add_avg("avg_latency")
                   .add_p95("p95_latency")
                   .add_p99("p99_latency");
                agg
            },
        );

    // Get results stream
    let mut results = processor.results().unwrap();

    // Process events
    processor.process_event(
        "my-service".to_string(),
        chrono::Utc::now(),
        123.45  // metric value
    ).await?;

    // Collect results
    tokio::spawn(async move {
        while let Some(result) = results.next().await {
            println!("Window: {}", result.window);
            println!("Results: {:?}", result.results);
        }
    });

    Ok(())
}

Window Types

Tumbling Windows

Fixed-size, non-overlapping windows for periodic aggregations:

let processor = StreamProcessorBuilder::new()
    .build_tumbling(Duration::minutes(5), aggregator_factory);

Sliding Windows

Overlapping windows for moving averages and smoothed metrics:

let processor = StreamProcessorBuilder::new()
    .build_sliding(
        Duration::minutes(10),  // Window size
        Duration::seconds(30),  // Slide interval
        aggregator_factory
    );

Session Windows

Gap-based windows for activity clustering:

let processor = StreamProcessorBuilder::new()
    .build_session(
        Duration::seconds(30),  // Inactivity gap
        aggregator_factory
    );

Aggregations

Composite Aggregator

Run multiple aggregations efficiently:

let mut agg = CompositeAggregator::new();
agg.add_count("total_requests")
   .add_avg("avg_response_time")
   .add_min("min_response_time")
   .add_max("max_response_time")
   .add_p50("p50_response_time")
   .add_p95("p95_response_time")
   .add_p99("p99_response_time")
   .add_stddev("stddev_response_time");

Simple Aggregators

Use individual aggregators for specific metrics:

use processor::aggregation::{
    AverageAggregator,
    PercentileAggregator,
    StandardDeviationAggregator,
};

let avg = AverageAggregator::new();
let p95 = PercentileAggregator::new(95);
let stddev = StandardDeviationAggregator::new();

Configuration

use processor::StreamProcessorConfig;
use chrono::Duration;

let config = StreamProcessorConfig {
    allow_late_events: true,
    late_event_threshold: Duration::seconds(30),
    watermark_interval: Duration::seconds(1),
    window_retention: Duration::hours(1),
    max_windows_per_key: 1000,
    result_buffer_size: 1000,
};

let processor = StreamProcessorBuilder::new()
    .with_config(config)
    .build_tumbling(Duration::minutes(5), aggregator_factory);

Advanced Statistics

The library includes advanced statistical utilities:

Online Variance

Numerically stable variance calculation:

use processor::aggregation::statistics::OnlineVariance;

let mut variance = OnlineVariance::new();
for value in values {
    variance.update(value);
}
let std_dev = variance.std_dev();

Exponential Moving Average

Smoothed metrics over time:

use processor::aggregation::statistics::ExponentialMovingAverage;

let mut ema = ExponentialMovingAverage::with_span(10.0)?;
ema.update(value);
let smoothed = ema.value();

Sliding Window Statistics

Fixed-size window with percentiles:

use processor::aggregation::statistics::SlidingWindowStats;

let mut stats = SlidingWindowStats::new(100)?;
for value in values {
    stats.update(value);
}
let p95 = stats.percentile(95.0)?;
let median = stats.median();

Anomaly Detection

Z-score based anomaly detection:

use processor::aggregation::statistics::ZScoreAnomalyDetector;

let mut detector = ZScoreAnomalyDetector::new(3.0); // 3-sigma threshold
let is_anomaly = detector.check(value);

Event Deduplication

Prevent duplicate events from being processed multiple times:

use processor::state::{RedisStateBackend, RedisConfig, StateBackend};
use std::sync::Arc;
use std::time::Duration;

// Create Redis backend for deduplication
let redis_config = RedisConfig::builder()
    .url("redis://localhost:6379")
    .key_prefix("dedup:")
    .default_ttl(Duration::from_secs(3600))
    .build()?;

let backend = Arc::new(RedisStateBackend::new(redis_config).await?);

// Check for duplicates
let event_id = "event_123";
let key = format!("dedup:{}", event_id);

if backend.get(key.as_bytes()).await?.is_none() {
    // First time seeing this event - process it
    backend.put_with_ttl(key.as_bytes(), b"1", Duration::from_secs(3600)).await?;
    process_event(event).await?;
} else {
    // Duplicate - skip processing
    info!("Duplicate event detected: {}", event_id);
}

Features

• Multiple Backends: Redis (distributed), Sled (local), In-memory
• Configurable TTL: Automatic cleanup of old deduplication records
• High Performance: 100K+ events/sec with Redis, 200K+ with Sled
• Bloom Filters: Probabilistic deduplication for ultra-high throughput
• Multi-Key Strategies: Event ID, transaction ID, idempotency keys, content hash

Configuration Options

// Basic configuration with in-memory backend
let backend = MemoryStateBackend::new(Some(Duration::from_secs(3600)));

// Production configuration with Redis
let redis_config = RedisConfig::builder()
    .url("redis://redis-cluster:6379")
    .key_prefix("dedup:")
    .default_ttl(Duration::from_secs(86400)) // 24 hours
    .pool_size(10, 50)  // min 10, max 50 connections
    .build()?;

// High-performance local storage with Sled
let sled_backend = SledStateBackend::new("./dedup.db")?;

Use Cases

1. Kafka Message Deduplication: Handle at-least-once delivery semantics
2. API Idempotency: Prevent duplicate requests using idempotency keys
3. Multi-Source Aggregation: Deduplicate events from multiple sources
4. Retry Safety: Ensure safe retries without duplicate processing
5. Exactly-Once Processing: Maintain exactly-once semantics across restarts

See deduplication/README.md for complete documentation.

Time-Series Normalization

Transform irregularly sampled time-series data into uniform intervals for analysis and processing:

use processor::normalization::{TimeSeriesNormalizer, NormalizationConfig, FillStrategy};
use chrono::Duration;

// Configure normalization with 1-second intervals
let config = NormalizationConfig::builder()
    .interval(Duration::seconds(1))
    .fill_strategy(FillStrategy::Linear)
    .max_gap_duration(Duration::seconds(60))
    .out_of_order_threshold(Duration::seconds(5))
    .build()?;

let mut normalizer = TimeSeriesNormalizer::new(config);

// Process irregular events
normalizer.process_event(timestamp1, value1).await?;
normalizer.process_event(timestamp2, value2).await?;

// Get normalized time-series with uniform intervals
let normalized = normalizer.normalize().await?;

Visual Example

Original (irregular sampling):
  t=0.0s: 10.0
  t=0.3s: 12.0
  t=1.7s: 18.0
  t=4.9s: 35.0

Normalized (1-second intervals, linear interpolation):
  t=0s: 10.0
  t=1s: 14.4   (interpolated)
  t=2s: 18.0
  t=3s: 24.3   (interpolated)
  t=4s: 30.7   (interpolated)
  t=5s: 35.0

Fill Strategies

• Linear: Interpolate between known points (best for continuous metrics)
• Forward Fill: Carry last value forward (best for discrete states)
• Backward Fill: Carry next value backward (batch processing)
• Zero Fill: Fill with zero (best for event counts)
• Mean Fill: Fill with average (best for noisy data)
• Polynomial: Higher-order interpolation (scientific data)
• Spline: Smooth curves without oscillation (high-quality visualization)

Features

• Irregular Sampling: Handle variable-rate data collection
• Out-of-Order Events: Automatic buffering and reordering
• Gap Filling: Intelligent interpolation for missing data
• Timestamp Alignment: Floor, ceil, or round to intervals
• Resampling: Upsample or downsample to target frequencies
• Multi-Metric: Normalize multiple metrics with different strategies
• Outlier Detection: Z-score based anomaly filtering
• High Performance: 500K+ events/sec with linear interpolation

Configuration Options

let config = NormalizationConfig {
    // Time interval between normalized points
    interval: Duration::seconds(1),

    // Timestamp alignment strategy
    alignment: AlignmentStrategy::Floor,  // or Ceil, Round

    // Gap filling strategy
    fill_strategy: FillStrategy::Linear,

    // Maximum gap to fill (larger gaps left as missing)
    max_gap_duration: Some(Duration::seconds(60)),

    // Out-of-order handling
    out_of_order_buffer_size: 1000,
    out_of_order_threshold: Duration::seconds(5),
};

Use Cases

1. IoT Sensor Data: Normalize irregular sensor readings for analysis
2. Financial Data: Align OHLC data to uniform intervals
3. System Metrics: Standardize monitoring data from multiple sources
4. ML Feature Engineering: Create uniform input for time-series models
5. Visualization: Generate smooth, consistent time-series plots
6. Aggregation Preparation: Align data before windowing operations

Performance Characteristics

See normalization/README.md for complete documentation.

Examples

Run the examples to see the stream processor in action:

# Basic stream processor demo
cargo run --example stream_processor_demo

# Multi-key sliding window
cargo run --example multi_key_sliding_window

# Event deduplication
cargo run --example event_deduplication_demo
cargo run --example advanced_deduplication

# Time-series normalization
cargo run --example timeseries_normalization_demo
cargo run --example advanced_normalization

# Other examples
cargo run --example kafka_pipeline
cargo run --example aggregation_demo
cargo run --example watermark_demo

Performance

Typical performance characteristics:

• Latency: 1-5ms processing time per event
• Throughput: 100,000+ events/sec (single key)
• Throughput: 1,000,000+ events/sec (multiple keys with parallelization)
• Memory: ~1KB per window + aggregation data

Monitoring

Get processor statistics:

let stats = processor.stats().await;
println!("Events processed: {}", stats.events_processed);
println!("Windows fired: {}", stats.windows_fired);
println!("Active windows: {}", stats.active_windows);
println!("P99 latency: {:.2}ms", stats.latency_p99_ms);

Architecture

See STREAM_PROCESSOR.md for detailed implementation documentation including:

• Architecture overview
• Implementation details
• Performance characteristics
• Tuning guidelines
• Best practices

Testing

# Run all tests
cargo test -p processor

# Run with output
cargo test -p processor -- --nocapture

# Run benchmarks
cargo bench -p processor

Integration

The stream processor integrates with:

• Kafka: Source and sink for event streaming
• State Backends: Redis, PostgreSQL, Sled for state management
• Watermark Generators: For event-time progress tracking
• Pipeline Framework: For complex stream processing workflows

License

Apache-2.0

Contributing

See CONTRIBUTING.md for contribution guidelines.