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';