DuckDB in Production: From Local Notebook to Enterprise Analytics Platform

# Increase resources on existing instance
resources:
 requests:
 memory: "32Gi"
 cpu: "8"
 limits:
 memory: "64Gi"
 cpu: "16"

# Multiple read replicas
apiVersion: apps/v1
kind: StatefulSet
metadata:
 name: duckdb-replica
spec:
 replicas: 3
 selector:
 matchLabels:
 app: duckdb-replica
 template:
 spec:
 containers:
 - name: duckdb
 image: duckdb:latest
 volumeMounts:
 - name: shared-storage
 mountPath: /data
 volumes:
 - name: shared-storage
 persistentVolumeClaim:
 claimName: duckdb-pvc

import duckdb
import time

class DuckDBHA:
 def __init__(self, primary_db, replica_db):
 self.primary_db = primary_db
 self.replica_db = replica_db
 self.current_db = primary_db
 self.con = None
 
 def connect(self):
 try:
 self.con = duckdb.connect(self.current_db)
 return True
 except Exception as e:
 self.failover()
 return False
 
 def failover(self):
 """Switch to replica if primary fails."""
 if self.current_db == self.primary_db:
 self.current_db = self.replica_db
 else:
 self.current_db = self.primary_db
 
 self.con = duckdb.connect(self.current_db)
 print(f"Failover to: {self.current_db}")
 
 def execute(self, query):
 try:
 return self.con.execute(query).fetchall()
 except Exception:
 self.failover()
 return self.con.execute(query).fetchall()

-- Enable WAL for crash recovery
PRAGMA enable_wal;

-- Configure checkpoint interval
PRAGMA checkpoint_threshold = 1000;

-- Backup to remote storage
COPY (SELECT * FROM orders) TO 's3://backup-bucket/orders_backup.parquet';

-- Restore from backup
ATTACH 's3://backup-bucket/orders_backup.parquet' AS backup (READ_ONLY);
SELECT * FROM backup.orders WHERE order_date > '2026-06-01';

from functools import wraps

class DuckDBSecurity:
 def __init__(self):
 self.roles = {
 'admin': ['SELECT', 'INSERT', 'UPDATE', 'DELETE', 'CREATE', 'DROP'],
 'analyst': ['SELECT'],
 'viewer': ['SELECT'],
 }
 self.user_roles = {}
 
 def authenticate(self, username, password):
 # In production, use proper authentication
 return True
 
 def authorize(self, username, operation):
 role = self.user_roles.get(username, 'viewer')
 allowed_ops = self.roles.get(role, [])
 return operation in allowed_ops
 
 def execute_with_security(self, username, query):
 if not self.authenticate(username, "password"):
 raise PermissionError("Authentication failed")
 
 # Extract operation from query
 operation = query.split()[0].upper()
 
 if not self.authorize(username, operation):
 raise PermissionError(f"Operation {operation} not allowed for role")
 
 return self.con.execute(query)

-- Enable encryption at rest
INSTALL encryption;
LOAD encryption;

-- Create encrypted database
ATTACH 'encrypted.db' AS encrypted (
 KEY 'your-encryption-key',
 TYPE encrypted
);

-- Query encrypted data
SELECT * FROM encrypted.orders;

import duckdb
import time
from datetime import datetime

class DuckDBMonitor:
 def __init__(self, db_path):
 self.db_path = db_path
 self.metrics = {
 'queries_per_minute': 0,
 'avg_query_time': 0,
 'memory_usage_mb': 0,
 'error_rate': 0,
 }
 
 def collect_metrics(self):
 con = duckdb.connect(self.db_path)
 
 # Query performance
 queries = con.execute("""
 SELECT 
 COUNT(*) as total_queries,
 AVG(execution_time_ms) as avg_time,
 MAX(execution_time_ms) as max_time,
 SUM(memory_used_mb) as total_memory
 FROM query_log
 WHERE timestamp > NOW() - INTERVAL '5 minutes'
 """).fetchone()
 
 self.metrics['queries_per_minute'] = queries[0] / 5
 self.metrics['avg_query_time'] = queries[1]
 self.metrics['memory_usage_mb'] = queries[3]
 
 # Error rate
 errors = con.execute("""
 SELECT COUNT(*) FROM query_log 
 WHERE status = 'error' 
 AND timestamp > NOW() - INTERVAL '5 minutes'
 """).fetchone()[0]
 
 self.metrics['error_rate'] = errors / max(queries[0], 1)
 
 con.close()
 return self.metrics
 
 def alert_thresholds(self):
 alerts = []
 
 if self.metrics['avg_query_time'] > 5000:
 alerts.append("High query latency detected")
 
 if self.metrics['memory_usage_mb'] > 8000:
 alerts.append("High memory usage detected")
 
 if self.metrics['error_rate'] > 0.05:
 alerts.append("High error rate detected")
 
 return alerts

{
 "dashboard": {
 "title": "DuckDB Production Monitor",
 "panels": [
 {
 "title": "Queries Per Minute",
 "type": "graph",
 "targets": [
 {
 "expr": "rate(duckdb_queries_total[5m])"
 }
 ]
 },
 {
 "title": "Query Latency",
 "type": "graph",
 "targets": [
 {
 "expr": "histogram_quantile(0.95, rate(duckdb_query_duration_seconds_bucket[5m]))"
 }
 ]
 },
 {
 "title": "Memory Usage",
 "type": "gauge",
 "targets": [
 {
 "expr": "duckdb_memory_used_bytes / duckdb_memory_limit_bytes"
 }
 ]
 },
 {
 "title": "Error Rate",
 "type": "stat",
 "targets": [
 {
 "expr": "rate(duckdb_errors_total[5m]) / rate(duckdb_queries_total[5m])"
 }
 ]
 }
 ]
 }
}

# test_duckdb.py
import duckdb
import pytest

def test_connection():
 con = duckdb.connect(':memory:')
 assert con is not None
 con.close()

def test_query_execution():
 con = duckdb.connect(':memory:')
 result = con.execute("SELECT 1 as test").fetchone()
 assert result[0] == 1
 con.close()

def test_performance():
 con = duckdb.connect(':memory:')
 con.execute("CREATE TABLE test AS SELECT * FROM generate_series(1, 1000000)")
 
 start = time.time()
 result = con.execute("SELECT COUNT(*) FROM test").fetchone()
 elapsed = time.time() - start
 
 assert result[0] == 1000000
 assert elapsed < 1.0 # Should complete in under 1 second
 con.close()

if __name__ == "__main__":
 pytest.main([__file__, "-v"])

# .github/workflows/deploy.yml
name: Deploy DuckDB Analytics

on:
 push:
 branches: [main]

jobs:
 test:
 runs-on: ubuntu-latest
 steps:
 - uses: actions/checkout@v3
 - name: Setup Python
 uses: actions/setup-python@v4
 with:
 python-version: '3.11'
 - name: Install dependencies
 run: pip install -r requirements.txt
 - name: Run tests
 run: pytest tests/

 deploy:
 needs: test
 runs-on: ubuntu-latest
 steps:
 - uses: actions/checkout@v3
 - name: Deploy to production
 run: |
 docker build -t duckdb-analytics:latest .
 docker push duckdb-analytics:latest
 kubectl rollout restart deployment/duckdb-analytics

-- Compress data efficiently
COPY data TO 's3://bucket/optimized/' (
 FORMAT PARQUET,
 COMPRESSION ZSTD,
 PER_THREAD_OUTPUT TRUE
);

-- Archive old data
COPY (
 SELECT * FROM orders 
 WHERE order_date < '2024-01-01'
) TO 's3://bucket/archive/old_orders/' (
 FORMAT PARQUET,
 COMPRESSION ZSTD
);

-- Set appropriate resource limits
SET memory_limit = '16GB';
SET max_threads = 8;
SET effective_io_concurrency = 200;

-- Use streaming for large queries
SET enable_streaming = TRUE;

-- Cache frequently accessed data
CREATE CACHE TABLE cached_recent_orders AS
SELECT * FROM orders WHERE order_date >= '2026-01-01';