DuckDB for Real-Time Analytics: What's Possible

import duckdb
import redis
import time
 
con = duckdb.connect('analytics.duckdb')
r = redis.Redis()
 
while True:
    # Pop batch of events from queue
    batch = []
    for _ in range(1000):  # Process up to 1000 events
        event = r.lpop('events')
        if not event:
            break
        batch.append(json.loads(event))
    
    if batch:
        # Batch insert into DuckDB
        con.executemany("""
            INSERT INTO events (user_id, event_type, value, occurred_at)
            VALUES (?, ?, ?, ?)
        """, [(e['user_id'], e['type'], e['value'], e['ts']) for e in batch])
    
    time.sleep(10)

-- Last hour activity
SELECT 
    DATE_TRUNC('minute', occurred_at) AS minute,
    COUNT(*) AS events,
    COUNT(DISTINCT user_id) AS active_users
FROM events
WHERE occurred_at > NOW() - INTERVAL '1 hour'
GROUP BY minute
ORDER BY minute;

import pandas as pd
import duckdb
from datetime import datetime
 
# Every 30 seconds: flush events buffer to a new Parquet file
def flush_events(events_buffer):
    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    df = pd.DataFrame(events_buffer)
    df.to_parquet(f'events/batch_{timestamp}.parquet')
    events_buffer.clear()

-- Query all recent files
SELECT 
    DATE_TRUNC('minute', occurred_at) AS minute,
    COUNT(*) AS events
FROM read_parquet('events/batch_*.parquet')
WHERE occurred_at > NOW() - INTERVAL '1 hour'
GROUP BY minute
ORDER BY minute;

-- Hot table: last 24 hours
CREATE TABLE events_hot AS 
SELECT * FROM events WHERE occurred_at > NOW() - INTERVAL '24 hours';
 
-- Cold archive: everything older (as Parquet)
COPY (
    SELECT * FROM events WHERE occurred_at <= NOW() - INTERVAL '24 hours'
) TO 'events_archive.parquet' (FORMAT PARQUET);
 
-- Query combining both
SELECT * FROM events_hot
UNION ALL
SELECT * FROM read_parquet('events_archive.parquet')
WHERE occurred_at > '2026-01-01';

// Frontend: poll every 30 seconds
async function refreshDashboard() {
    const response = await fetch('/api/analytics/current');
    const data = await response.json();
    updateCharts(data);
}
 
setInterval(refreshDashboard, 30000);
refreshDashboard(); // Initial load

# Backend: query DuckDB on demand
from flask import Flask, jsonify
import duckdb
 
app = Flask(__name__)
con = duckdb.connect('analytics.duckdb', read_only=True)
 
@app.route('/api/analytics/current')
def current_analytics():
    result = con.execute("""
        SELECT 
            COUNT(*) AS events_last_hour,
            COUNT(DISTINCT user_id) AS active_users,
            SUM(CASE WHEN event_type = 'purchase' THEN value ELSE 0 END) AS revenue_today
        FROM events
        WHERE occurred_at > NOW() - INTERVAL '1 hour'
    """).fetchone()
    
    return jsonify({
        'events': result[0],
        'active_users': result[1],
        'revenue': float(result[2])
    })

#!/usr/bin/env python3
# alert_check.py - run every minute via cron
 
import duckdb
import requests
 
con = duckdb.connect('analytics.duckdb', read_only=True)
 
# Check for error rate spike
error_rate = con.execute("""
    SELECT 
        COUNT(CASE WHEN event_type = 'error' THEN 1 END) * 100.0 / COUNT(*) AS error_rate
    FROM events
    WHERE occurred_at > NOW() - INTERVAL '5 minutes'
""").fetchone()[0]
 
if error_rate > 5.0:  # Alert if error rate > 5%
    requests.post('https://hooks.slack.com/services/...', json={
        'text': f'⚠️ Error rate spike: {error_rate:.1f}% in last 5 minutes'
    })

# crontab: run every minute
* * * * * /path/to/alert_check.py

// DenchClaw app with auto-refresh
async function loadMetrics() {
    const metrics = await dench.db.query(`
        SELECT 
            "Stage",
            COUNT(*) AS deals,
            SUM("Deal Value") AS value
        FROM v_deals
        WHERE "Status" = 'Active'
        GROUP BY "Stage"
    `);
    updatePipelineChart(metrics);
}
 
// Refresh every 60 seconds
setInterval(loadMetrics, 60000);
loadMetrics();