Data Pipeline — Automazione Dati PMI
Architettura pipeline PMI standard
SORGENTI INGESTIONE STORAGE CONSUMO
───────── ────────── ─────── ───────
Gestionale ──→ ETL (n8n/ Supabase ──→ Dashboard
ERP ──→ Make/custom) ──→ PostgreSQL ──→ Report PDF
Fogli Google──→ Data Warehouse──→ Alert email
CRM ──→ (BigQuery) ──→ AI/RAG
Form/webhook──→ ──→ Esportazione
ETL — fasi
Extract
- REST API polling (ogni 15min/1h/24h per volume)
- File drop (CSV/XLSX in cartella → trigger)
- Webhook push (real-time)
- Database replication (Postgres CDC via Supabase realtime)
Transform — operazioni comuni
| Operazione |
Esempio |
| Normalizza date |
dd/mm/yyyy → ISO 8601 yyyy-mm-dd |
| Converti importi |
Stringa "€ 1.234,56" → float 1234.56 |
| Deduplication |
Upsert su chiave codice_cliente + data |
| Lookup codici |
Codice prodotto → categoria merceologica |
| Calcola metriche |
margine = ricavi - costi_diretti |
| Enrich |
Aggiungi ATECO, CAP → regione, P.IVA → ragione sociale |
Load
- Upsert su Supabase (PostgreSQL):
ON CONFLICT DO UPDATE
- Append-only per log/eventi (mai modificare storico)
- Soft delete:
deleted_at timestamp, non DELETE fisico
Qualità dati — regole obbligatorie
Completezza: campi obbligatori presenti > 95%
Unicità: no duplicati su chiave primaria
Validità: P.IVA 11 cifre, CF 16 car., IBAN IT27+18
Consistenza: data_fine ≥ data_inizio, importo > 0
Freschezza: dati core aggiornati entro SLA (es. 1h per vendite)
Implementa check con SQL assertion o n8n IF node:
-- Esempio quality check giornaliero
SELECT COUNT(*) as righe_problematiche
FROM ordini
WHERE cliente_id IS NULL
OR importo <= 0
OR data_ordine > CURRENT_DATE
Scheduling — criteri di frequenza
| Tipo dato |
Frequenza |
Motivo |
| Vendite/ordini |
15-30 min |
Decisioni operative intraday |
| Magazzino/stock |
1h |
Bilanciamento scorte |
| Fatturazione |
2h |
Riconciliazione pagamenti |
| HR / presenze |
Giornaliera (06:00) |
Elaborazioni batch notturne |
| Bilancio |
Settimanale |
Consolidamento dati |
| KPI strategici |
Mensile |
Reporting management |
Pattern Supabase PostgreSQL per PMI
-- Tabella vendite denormalizzata per velocità query
CREATE TABLE vendite_agg (
id BIGSERIAL PRIMARY KEY,
data DATE NOT NULL,
cliente_id TEXT NOT NULL,
prodotto TEXT,
quantita NUMERIC(12,3),
importo_netto NUMERIC(12,2),
margine NUMERIC(12,2),
agente TEXT,
created_at TIMESTAMPTZ DEFAULT NOW(),
source TEXT -- 'gestionale', 'ecommerce', 'manuale'
);
CREATE INDEX ON vendite_agg(data, cliente_id);
CREATE INDEX ON vendite_agg(data, agente);
Gestione errori pipeline
- Dead letter queue: righe fallite → tabella
etl_errors(ts, source, row_data, error_msg)
- Retry policy: max 3 tentativi con backoff 1min/5min/30min
- Alert soglia: se
etl_errors > 10 in 1h → email alert
- Idempotenza: ogni run può girare 2x senza doppi inserimenti (upsert, non insert)
Stack raccomandato K2-AI per PMI
Orchestration: n8n self-host (Railway, ~15€/mese) o Make (free 1000 ops)
Storage: Supabase EU Frankfurt (free 500MB, poi 25€/mese)
Trasformazioni: JavaScript inline n8n o Python script su Railway
Dashboard: Supabase + Metabase (self-host) o Google Looker Studio
Alerting: Resend email + webhook Slack/Teams
Costo totale: 20-50€/mese per PMI < 50 dipendenti
Anti-pattern da evitare
- Mai SELECT * in produzione: specifica sempre le colonne
- No logica business nel DB: mantienila nell'ETL, non in trigger SQL
- Non eliminare raw data: archivia sempre l'originale prima di trasformare
- No credenziali hardcoded: sempre env vars, mai in codice o n8n variabili hardcoded
gigik2a