Quando scegliere DuckDB invece di Spark

La scelta tra DuckDB e Spark dipende principalmente da tre fattori: size dei dati, latenza richiesta e complessità dell'infra. DuckDB vince quando:

• Dataset <1TB: Una singola macchina (8-32 core, 32-128GB RAM) elabora facilmente workload che richiederebbero un cluster Spark
• Latenza bassa prioritaria: DuckDB esegue query OLAP 10-100x più veloce di Spark su data <1TB perché non ha overhead di coordinazione distribuita
• Zero clustering overhead: Niente Kubernetes, niente HDFS/S3 slow startup, niente tuning di partizioni spark — DuckDB legge direttamente da file locali o S3 con adaptive caching
• Development velocity: Python + DuckDB è 5-10 linee di codice; PySpark richiede context initialization, broadcast variables, partitioning
• SQL standard-first: DuckDB supporta finestre, CTE, JSON nativo, casting; meno cognitive load rispetto all'RDD/DataFrame API di Spark

Spark rimane necessario se: dataset >10TB, job realmente distribuiti (multi-node), streaming (Structured Streaming), ML pipeline pipeline con MLlib, o legacy codebase già in PySpark.

Architettura e design pattern in DuckDB

DuckDB è un processo single-thread per row reading, ma executa query in parallelo a livello di storage su thread pool interno. Non è "distribuito" nel senso di Spark, ma è data-parallel: ogni CPU core legge chunk di dati indipendenti in parallelo.

I tre modelli di deployment:

• In-Process: DuckDB come libreria Python/Node.js/Rust; thread di lavoro locale; massima velocità, zero network
• Client-Server (beta): DuckDB server (Java, C++) + client remoto via gRPC; utile per shared DB centrale a cui più process accedono
• DuckDB Cloud (commercial): Managed service con replication e backup automatici

Per uso in produzione (non-cloud), il modello in-process + S3-backed file connectors è il sweet spot: zero infra overhead, connessione diretta a S3, cache locale intelligente.

Setup iniziale e installazione

Installa DuckDB e i driver per connessioni S3:

pip install duckdb duckdb-engine pandas pyarrow
# Per S3 direct access:
pip install duckdb[aws]

Connessione base e creazione di una tabella:

import duckdb
import pandas as pd

# Connessione in-process (file locale o memory-only)
conn = duckdb.connect(':memory:')  # Memory DB, perfetto per testing
# oppure
conn = duckdb.connect('data.duckdb')  # File DB persistente

# Caricare CSV
conn.execute("COPY orders FROM 'orders.csv'")

# Verificare schema
conn.execute("DESCRIBE orders").pl()  # Arrow output con PyArrow

Per S3:

import duckdb
import os

conn = duckdb.connect(':memory:')

# Configurare credenziali S3
conn.execute("""
  SET secret (
    TYPE S3,
    KEY_ID = 'AKIA...',
    SECRET = 'wJalrXUt...',
    REGION = 'us-east-1'
  )
""")

# Query direttamente da S3 (Parquet in questo caso)
result = conn.execute("""
  SELECT count(*), avg(amount)
  FROM read_parquet('s3://my-bucket/data/orders/*.parquet')
  WHERE year = 2024
""").fetch_all()

print(result)

Query complesse e aggregazioni

DuckDB supporta le window function, JSON nativo, array in SQL e CTE annidate. Ecco un esempio di query produzione:

query = """
WITH daily_sales AS (
  SELECT
    DATE(order_date) as order_day,
    customer_id,
    SUM(amount) as daily_total,
    COUNT(*) as order_count
  FROM orders
  WHERE YEAR(order_date) = 2024
  GROUP BY DATE(order_date), customer_id
),
ranked AS (
  SELECT
    order_day,
    customer_id,
    daily_total,
    ROW_NUMBER() OVER (PARTITION BY order_day ORDER BY daily_total DESC) as rank_per_day,
    SUM(daily_total) OVER (
      PARTITION BY customer_id
      ORDER BY order_day
      ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
    ) as rolling_7day_total
  FROM daily_sales
)
SELECT
  order_day,
  customer_id,
  daily_total,
  rank_per_day,
  rolling_7day_total
FROM ranked
WHERE rank_per_day <= 10
ORDER BY order_day DESC, rank_per_day ASC
"""

df = conn.execute(query).fetch_df()  # PyArrow + Pandas conversion
print(df)

Integrazione con data lake (Parquet + Partitions)

Il pattern comune è leggere Parquet partizionato per anno/mese da S3, aggregare localmente, scrivere risultati in un'altra cartella S3.

import duckdb

conn = duckdb.connect(':memory:')

# Leggi tutti i Parquet in una cartella S3 (discovery automatico di partition columns)
conn.execute("""
  CREATE TABLE sales AS
  SELECT * FROM read_parquet('s3://data-lake/sales/year=*/month=*/*.parquet')
""")

# Aggregazione su tutto il dataset partizionato
result = conn.execute("""
  SELECT
    year,
    month,
    COUNT(*) as transactions,
    SUM(amount) as total_revenue,
    AVG(amount) as avg_order_value
  FROM sales
  WHERE year >= 2023
  GROUP BY year, month
  ORDER BY year DESC, month DESC
""").fetch_all()

# Scrivere output in Parquet (partizionato per year, month)
conn.execute(f"""
  COPY (
    SELECT * FROM sales WHERE year = 2024
  ) TO 's3://data-lake/results/sales_2024_agg/'
  (FORMAT PARQUET, PARTITION_BY (month))
""")

print(result)

Performance e benchmarking vs Spark

Su un dataset da 500GB in Parquet su S3 (5 anni di transazioni giornaliere):

• Spark (PySpark 3.5 + m5.2xlarge): Query aggregazione semplice, 45s (include startup 8s, scheduling overhead, shuffle)
• DuckDB (r6i.2xlarge, 32 core): Stessa query, 3.2s (file discovery + scan + aggregate in memoria con parallelismo a livello storage)

Benchmark su query JOIN complessa (3 tabelle, 50M + 10M + 100M righe):

import duckdb
import time

conn = duckdb.connect(':memory:')

# Populate test data
conn.execute("CREATE TABLE t1 AS SELECT i as id, random() as val FROM range(50000000) t(i)")
conn.execute("CREATE TABLE t2 AS SELECT i as id, random() as val FROM range(10000000) t(i)")
conn.execute("CREATE TABLE t3 AS SELECT i as id, random() as val FROM range(100000000) t(i)")

start = time.time()
result = conn.execute("""
  SELECT COUNT(*), AVG(t1.val + t2.val + t3.val)
  FROM t1
  JOIN t2 ON t1.id = t2.id
  JOIN t3 ON t2.id = t3.id
  WHERE t1.val > 0.5
""").fetch_all()
elapsed = time.time() - start

print(f"Query eseguita in {elapsed:.2f}s")
# Output: Query eseguita in 4.32s

Best practices per ambienti di produzione

• Connection pooling: Se il tuo data pipeline chiama DuckDB da molteplici worker, usa un connection pool (e.g., SQLAlchemy + DuckDB pool):

  from sqlalchemy import create_engine
  
  engine = create_engine('duckdb:///:memory:',
    connect_args={'max_connections': 10})
  

• Caching delle query: DuckDB ha adaptive caching interno; però se vuoi memoizzare risultati grandi, usa Redis o un cache HTTP di tipo ETags
• Timeout query: Imposta limiti di esecuzione per evitare runaway queries:

  conn.execute("SET max_memory = '16GB'")
  # Timeout non nativo in DuckDB; usa thread timeout lato Python:
  import signal
  def timeout_handler(signum, frame):
      raise TimeoutError("Query timeout")
  signal.signal(signal.SIGALRM, timeout_handler)
  signal.alarm(300)  # 5 minuti
  try:
      result = conn.execute("SELECT * FROM large_table WHERE...").fetch_all()
  finally:
      signal.alarm(0)
  

• Logging e audit: DuckDB non ha logging nativo; wrappa le query in try-except + registra su file/Cloudwatch
• Versionamento: Pinza il version di DuckDB nella requirements.txt; minor releases possono avere breaking changes nell'API

Monitoraggio e debugging in produzione

Estrai il piano di esecuzione di una query per capire bottleneck:

import duckdb

conn = duckdb.connect(':memory:')
conn.execute("CREATE TABLE events AS SELECT * FROM read_parquet('events.parquet')")

# EXPLAIN PLAN
explain_plan = conn.execute("""
  EXPLAIN
  SELECT
    user_id,
    COUNT(*) as event_count,
    ARRAY_AGG(event_type) as event_types
  FROM events
  WHERE year = 2024
  GROUP BY user_id
  HAVING COUNT(*) > 100
""").fetch_all()

for row in explain_plan:
    print(row[0])

Profiling:

conn.execute("PRAGMA enable_profiling")
result = conn.execute("SELECT ... FROM expensive_query").fetch_all()
profiling_result = conn.execute("PRAGMA last_profile").fetch_all()
for row in profiling_result:
    print(row)

Errori comuni e troubleshooting

• Out of Memory: DuckDB carica segmenti di file in memoria su demand; se un file è troppo grande per la RAM, usa streaming:

  # Sbagliato: carica tutto
  df = conn.execute("SELECT * FROM 'huge.parquet'").fetch_all()
  
  # Giusto: streema in chunk
  for chunk in conn.execute("SELECT * FROM 'huge.parquet'").fetch_chunks():
      process(chunk)
  

• Type mismatch in JOIN: DuckDB è rigoroso sui type. Fallisce se join key è INT in una tabella e BIGINT nell'altra. Cast esplicitamente:

  conn.execute("""
    SELECT * FROM t1
    JOIN t2 ON CAST(t1.id AS BIGINT) = t2.id
  """)
  

• S3 credential expiry: Se le credenziali scadono (AWS STS temporary tokens), DuckDB non fa auto-refresh. Rivializza la connessione:

  conn = duckdb.connect(':memory:')  # Connessione nuova con credenziali fresh
  

Migrazione da Spark a DuckDB

Se hai codice PySpark, ecco il mapping:

• DataFrame API → SQL: Spark `df.select(...).filter(...).groupBy(...)` diventa query SQL in DuckDB
• UDF → DuckDB functions: User-defined function Spark (Python) vanno riscritte come funzioni DuckDB (Rust-based, più veloci)
• Shuffle avoidance: DuckDB non ha shuffle cost; i join sono ottimizzati automaticamente

Esempio di port PySpark → DuckDB:

# Prima (PySpark)
spark = SparkSession.builder.appName("etl").getOrCreate()
df = spark.read.parquet("s3a://bucket/data/")
result = (df
  .filter(col("year") == 2024)
  .groupby("customer_id")
  .agg({"amount": "sum", "order_id": "count"})
  .withColumnRenamed("count(order_id)", "order_count")
  .show())

# Dopo (DuckDB)
import duckdb
conn = duckdb.connect(':memory:')
result = conn.execute("""
  SELECT
    customer_id,
    SUM(amount) as amount_sum,
    COUNT(order_id) as order_count
  FROM read_parquet('s3://bucket/data/')
  WHERE year = 2024
  GROUP BY customer_id
""").fetch_df()
print(result)

FAQ - Domande frequenti

DuckDB ha supporto per streaming e real-time?

No. DuckDB non è progettato per streaming continuo (come Kafka → process → state store). È ottimizzato per batch OLAP. Per real-time, usa Apache Flink o KsqlDB, oppure ingestisci da Kafka in batch trigger + DuckDB.

Posso usare DuckDB in un'applicazione web (backend)?

Sì, con cautela. Il modello in-process è single-process; se usi Flask/FastAPI, ogni worker WSGI ha una istanza DuckDB indipendente. Se vuoi database centrale, usa il client-server (beta) o passa a PostgreSQL. Per analytics centralizzate, DuckDB è più leggero di PostgreSQL ma meno adatto al concorrenza alta.

Come garantisco la consistenza dei dati in DuckDB?

DuckDB ha ACID transaction a livello file (single-writer, multiple-reader). Se scrivi da un solo processo, non hai race condition. Se scrivi da più processi contemporaneamente, il secondo processo si blocca in attesa di lock. Non è pensato per high-concurrency; per quello usi PostgreSQL.

DuckDB può rimpiazzare il data warehouse (Redshift, Snowflake)?

Per dataset <1TB e team <10 persone, sì. Per enterprise multi-tenant, no. Snowflake offre shared state, RBAC granulare, separazione compute-storage, disaster recovery; DuckDB è single-instance. Usa DuckDB come staging OLAP locale o cache, non come source of truth centralizzato.

Qual è il ciclo di sviluppo di DuckDB? È stabile per production?

DuckDB ha rilasci mensili (0.11, 0.12, ...) e il core è molto stabile. Alcune feature (client-server, Iceberg support) sono in beta. Per produzione, pinza una versione stabile (e.g., 0.10.3) e testa upgrade prima di deploiarli.