Real-Time Streaming Analytics

Problem Being Solved

Root Cause: Traditional databases process batch queries; streaming requires separate infrastructure:

• System Sprawl: Kafka + Flink + PostgreSQL + Redis + API = 5 systems to maintain
• Data Duplication: Same data in message queue, processing engine, and database
• Consistency Gaps: Eventually consistent between stream and database
• Operational Complexity: Different skills for each system

HeliosDB Solution: Unified Streaming

Streaming SQL Interface

-- Create a stream from Kafka topic
CREATE STREAM clickstream (
    user_id UUID,
    page_url VARCHAR(500),
    event_type VARCHAR(50),
    event_time TIMESTAMPTZ,
    session_id UUID
) WITH (
    connector = 'kafka',
    topic = 'user-clicks',
    format = 'json',
    timestamp_column = 'event_time'
);

-- Real-time page view aggregation (tumbling window)
CREATE MATERIALIZED VIEW page_views_per_minute AS
SELECT
    page_url,
    COUNT(*) as view_count,
    COUNT(DISTINCT user_id) as unique_visitors,
    WINDOW_START as minute_start,
    WINDOW_END as minute_end
FROM clickstream
GROUP BY
    page_url,
    TUMBLE(event_time, INTERVAL '1 minute');

-- Query the materialized view (always up-to-date)
SELECT * FROM page_views_per_minute
WHERE minute_start > NOW() - INTERVAL '5 minutes'
ORDER BY view_count DESC
LIMIT 10;

-- Stream-to-stream join (sessionization)
CREATE STREAM user_sessions AS
SELECT
    c.user_id,
    c.session_id,
    MIN(c.event_time) as session_start,
    MAX(c.event_time) as session_end,
    COUNT(*) as event_count,
    ARRAY_AGG(c.page_url ORDER BY c.event_time) as page_sequence
FROM clickstream c
GROUP BY
    c.user_id,
    c.session_id,
    SESSION(c.event_time, INTERVAL '30 minutes');

Stream-Table Joins

-- Join stream with static table for enrichment
CREATE STREAM enriched_clicks AS
SELECT
    c.user_id,
    c.page_url,
    c.event_time,
    u.subscription_tier,
    u.country,
    p.product_category
FROM clickstream c
JOIN users u ON c.user_id = u.id
LEFT JOIN products p ON c.page_url LIKE '%/product/' || p.id || '%';

-- Real-time fraud detection with table lookup
CREATE STREAM fraud_alerts AS
SELECT
    t.transaction_id,
    t.user_id,
    t.amount,
    t.merchant,
    'HIGH_RISK' as alert_type,
    CURRENT_TIMESTAMP as alert_time
FROM transactions_stream t
JOIN user_risk_profiles r ON t.user_id = r.user_id
WHERE t.amount > r.typical_amount * 3
   OR t.merchant_country != r.home_country
   OR t.hour_of_day NOT BETWEEN r.typical_start_hour AND r.typical_end_hour;

Concrete Examples

Example 1: E-commerce Real-Time Analytics

Company Profile: Online retailer with 10M daily visitors

Challenge: Real-time conversion funnel analysis

-- Create streams for user events
CREATE STREAM user_events (
    user_id UUID,
    event_type VARCHAR(50),  -- 'page_view', 'add_to_cart', 'checkout', 'purchase'
    product_id UUID,
    amount DECIMAL(10,2),
    event_time TIMESTAMPTZ
) WITH (connector = 'kafka', topic = 'user-events');

-- Real-time conversion funnel
CREATE MATERIALIZED VIEW conversion_funnel_5min AS
SELECT
    WINDOW_START as time_bucket,
    COUNT(DISTINCT CASE WHEN event_type = 'page_view' THEN user_id END) as visitors,
    COUNT(DISTINCT CASE WHEN event_type = 'add_to_cart' THEN user_id END) as cart_adds,
    COUNT(DISTINCT CASE WHEN event_type = 'checkout' THEN user_id END) as checkouts,
    COUNT(DISTINCT CASE WHEN event_type = 'purchase' THEN user_id END) as purchases,
    ROUND(
        COUNT(DISTINCT CASE WHEN event_type = 'purchase' THEN user_id END)::FLOAT /
        NULLIF(COUNT(DISTINCT CASE WHEN event_type = 'page_view' THEN user_id END), 0) * 100,
        2
    ) as conversion_rate
FROM user_events
GROUP BY TUMBLE(event_time, INTERVAL '5 minutes');

-- Real-time revenue by product category
CREATE MATERIALIZED VIEW revenue_by_category_hourly AS
SELECT
    p.category,
    SUM(e.amount) as revenue,
    COUNT(*) as order_count,
    AVG(e.amount) as avg_order_value,
    WINDOW_START as hour_start
FROM user_events e
JOIN products p ON e.product_id = p.id
WHERE e.event_type = 'purchase'
GROUP BY p.category, TUMBLE(e.event_time, INTERVAL '1 hour');

Example 2: IoT Sensor Monitoring

Company Profile: Smart building management for 500 buildings

Challenge: Real-time anomaly detection across 1M sensors

-- Create sensor data stream
CREATE STREAM sensor_readings (
    building_id VARCHAR(50),
    sensor_id VARCHAR(100),
    sensor_type VARCHAR(50),
    reading FLOAT,
    reading_time TIMESTAMPTZ
) WITH (
    connector = 'mqtt',
    topic = 'sensors/#',
    throughput = '100000/sec'
);

-- Rolling statistics for anomaly baseline
CREATE MATERIALIZED VIEW sensor_baselines AS
SELECT
    sensor_id,
    sensor_type,
    AVG(reading) as avg_reading,
    STDDEV(reading) as stddev_reading,
    MIN(reading) as min_reading,
    MAX(reading) as max_reading,
    COUNT(*) as sample_count
FROM sensor_readings
GROUP BY sensor_id, sensor_type, HOP(reading_time, INTERVAL '1 minute', INTERVAL '1 hour');

-- Real-time anomaly detection stream
CREATE STREAM sensor_anomalies AS
SELECT
    s.sensor_id,
    s.building_id,
    s.sensor_type,
    s.reading,
    s.reading_time,
    b.avg_reading,
    b.stddev_reading,
    ABS(s.reading - b.avg_reading) / NULLIF(b.stddev_reading, 0) as z_score,
    CASE
        WHEN ABS(s.reading - b.avg_reading) > 3 * b.stddev_reading THEN 'CRITICAL'
        WHEN ABS(s.reading - b.avg_reading) > 2 * b.stddev_reading THEN 'WARNING'
        ELSE 'NORMAL'
    END as severity
FROM sensor_readings s
JOIN sensor_baselines b ON s.sensor_id = b.sensor_id
WHERE ABS(s.reading - b.avg_reading) > 2 * b.stddev_reading;

-- Automated alerting
CREATE SINK alert_sink AS
SELECT * FROM sensor_anomalies
WHERE severity IN ('CRITICAL', 'WARNING')
WITH (
    connector = 'webhook',
    url = 'https://alerts.company.com/sensor',
    batch_size = 1,
    latency = '<1s'
);

Example 3: Financial Transaction Processing

-- Create transaction stream
CREATE STREAM transactions (
    transaction_id UUID,
    card_number_hash VARCHAR(64),
    merchant_id VARCHAR(50),
    amount DECIMAL(15,2),
    currency VARCHAR(3),
    country VARCHAR(2),
    transaction_time TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'transactions',
    exactly_once = true
);

-- Real-time velocity checks (sliding window)
CREATE MATERIALIZED VIEW card_velocity AS
SELECT
    card_number_hash,
    COUNT(*) as txn_count_1h,
    SUM(amount) as total_amount_1h,
    COUNT(DISTINCT merchant_id) as unique_merchants_1h,
    COUNT(DISTINCT country) as unique_countries_1h
FROM transactions
GROUP BY
    card_number_hash,
    HOP(transaction_time, INTERVAL '1 minute', INTERVAL '1 hour');

-- Real-time fraud scoring
CREATE STREAM fraud_scores AS
SELECT
    t.transaction_id,
    t.card_number_hash,
    t.amount,
    t.merchant_id,
    -- Velocity-based risk factors
    CASE WHEN v.txn_count_1h > 10 THEN 0.3 ELSE 0 END +
    CASE WHEN v.unique_countries_1h > 2 THEN 0.4 ELSE 0 END +
    CASE WHEN t.amount > v.total_amount_1h THEN 0.2 ELSE 0 END +
    -- Pattern-based risk
    CASE WHEN t.amount = ROUND(t.amount) AND t.amount > 100 THEN 0.1 ELSE 0 END
    as fraud_score,
    t.transaction_time
FROM transactions t
LEFT JOIN card_velocity v ON t.card_number_hash = v.card_number_hash;

-- Block high-risk transactions in real-time
CREATE SINK fraud_blocks AS
SELECT
    transaction_id,
    fraud_score,
    'BLOCKED' as decision
FROM fraud_scores
WHERE fraud_score > 0.7
WITH (
    connector = 'kafka',
    topic = 'fraud-decisions',
    latency_sla = '50ms'
);

Example 4: Log Analytics & Observability

Company Profile: SaaS platform with 1000 microservices

Challenge: Real-time log analysis and alerting

-- Create log stream
CREATE STREAM application_logs (
    service_name VARCHAR(100),
    instance_id VARCHAR(50),
    log_level VARCHAR(10),
    message TEXT,
    trace_id UUID,
    log_time TIMESTAMPTZ
) WITH (
    connector = 'kafka',
    topic = 'app-logs',
    format = 'json'
);

-- Error rate monitoring by service (1-minute windows)
CREATE MATERIALIZED VIEW service_error_rates AS
SELECT
    service_name,
    WINDOW_START as minute,
    COUNT(*) as total_logs,
    COUNT(*) FILTER (WHERE log_level = 'ERROR') as error_count,
    COUNT(*) FILTER (WHERE log_level = 'WARN') as warn_count,
    ROUND(
        COUNT(*) FILTER (WHERE log_level = 'ERROR')::FLOAT /
        NULLIF(COUNT(*), 0) * 100, 2
    ) as error_rate_pct
FROM application_logs
GROUP BY service_name, TUMBLE(log_time, INTERVAL '1 minute');

-- Automatic incident detection
CREATE STREAM incidents AS
SELECT
    service_name,
    minute,
    error_rate_pct,
    'ERROR_SPIKE' as incident_type,
    CURRENT_TIMESTAMP as detected_at
FROM service_error_rates
WHERE error_rate_pct > 5  -- More than 5% errors
   OR error_count > 100;   -- Or absolute threshold

-- Trace aggregation for debugging
CREATE MATERIALIZED VIEW trace_summary AS
SELECT
    trace_id,
    ARRAY_AGG(service_name ORDER BY log_time) as service_sequence,
    MIN(log_time) as trace_start,
    MAX(log_time) as trace_end,
    MAX(log_time) - MIN(log_time) as trace_duration,
    COUNT(*) FILTER (WHERE log_level = 'ERROR') as error_count
FROM application_logs
WHERE trace_id IS NOT NULL
GROUP BY trace_id, SESSION(log_time, INTERVAL '5 minutes');

Conclusion

HeliosDB's unified streaming analytics eliminates the traditional separation between stream processing and databases. Organizations can:

• Simplify Architecture: One system instead of 3-5
• Ensure Consistency: ACID guarantees on streaming data