Le tre lezioni precedenti hanno percorso la meccanica del partizionare un database su molte macchine. Hash sharding, range sharding, il problema del rebalancing, il problema delle hot key. Alla fine della lezione 30 avevi un cluster shardato e una shard key difendibile. Questa lezione parla della conseguenza: ora che i tuoi dati sono distribuiti su N macchine, ogni query che fai ha un costo che dipende da quanti shard deve toccare, e l’intera architettura è costruita attorno al tenere quel numero il più piccolo possibile. Di solito uno.
Il fatto di base, detto senza giri di parole: un database shardato è veloce sulle query a uno shard e lento sulle query a N shard. Una query che atterra su un singolo shard gira alla velocità di un singolo shard, che su Postgres o MySQL è un millisecondo o due per una lookup su primary key. Una query che deve fare fan-out a tutti gli shard gira alla velocità dello shard più lento, più l’overhead di mergiare i risultati. Se gli shard sono sulla stessa rete, quel fan-out è di pochi millisecondi a una cifra. Se sono sparsi tra region diverse, sono centinaia. In ogni caso, scala peggio di una query a uno shard, e man mano che il cluster cresce peggiora, non migliora.
Tutto il gioco del progettare un sistema shardato è tenere le query su un solo shard ogni volta che è possibile. Le due strategie per farlo sono co-location e fan-out, e sono complementari, non concorrenti.
Co-location: disponi i dati in modo che la query stia su un solo shard
La co-location è la strategia proattiva. Disponi il layout dei tuoi dati in modo che le righe che una query tipica deve guardare insieme vivano sullo stesso shard. La shard key è la leva. Sceglila in modo che le entità che vengono interrogate insieme condividano una chiave.
La scelta classica e quasi universale per le applicazioni SaaS è shardare per user_id. Il profilo di un utente, i suoi ordini, le sue impostazioni, le sue sessioni, le sue notifiche, i suoi file caricati: tutto ottiene l’ID dell’utente come parte della shard key, il che significa che tutto vive sullo stesso shard. Qualsiasi query che filtri per user_id (cioè quasi ogni query in un’app user-centric) colpisce esattamente uno shard.
Questo funziona perché i workload user-centric hanno una proprietà che si allinea perfettamente con il modello di sharding. La maggior parte delle query è circoscritta a un singolo utente. I miei ordini sono indipendenti dai tuoi: non c’è una query normale che debba fare join tra i miei ordini e i tuoi ordini. I dati hanno un partizionamento naturale lungo l’asse user_id, e il pattern di accesso dell’applicazione lo rispetta.
Lo stesso ragionamento si generalizza. Un SaaS B2B multi-tenant sharda per tenant_id invece che per user_id, perché l’unità naturale di località è il tenant, non il singolo utente. Un’app di messaggistica sharda per channel_id (il caso di studio di Discord nella prossima lezione è esattamente questo). Un gioco social sharda per world_id o instance_id. Il pattern è lo stesso: identifica l’entità che possiede la maggior parte dei dati e la maggior parte delle query, e usa il suo ID come shard key.
Quando azzecchi la shard key, il sistema sembra quasi un database non shardato. Le query sono veloci, la latenza è prevedibile, il cluster scala linearmente con il numero di utenti (o tenant, o canali). Il fatto che ci siano cinquanta macchine sotto il cofano è invisibile al codice applicativo, che semplicemente filtra per user_id e ottiene performance da uno shard ogni volta.
Fan-out: quando davvero non puoi evitare di toccare ogni shard
Alcune query non possono essere fatte vivere su un solo shard, per quanto astuta sia la tua shard key. “Quanti utenti hanno fatto login oggi sull’intera piattaforma” non si può rispondere da uno shard, perché i login sono distribuiti su tutti per user_id. “Mostrami tutti gli ordini piazzati nell’ultima ora, ordinati per valore” non si può rispondere da uno shard per la stessa ragione. Qualsiasi cosa aggreghi tra utenti, o filtri per un campo che non sia la shard key, o faccia ordinamento globale, deve guardare ogni shard.
Per queste query la strategia è il fan-out. Il coordinatore della query (a volte l’applicazione, a volte un router middle-tier, a volte il database stesso) manda la query a ogni shard in parallelo. Ogni shard esegue la sua parte localmente e restituisce il suo risultato. Il coordinatore mergia i risultati parziali nella risposta finale.
Il modello di costo per il fan-out è semplice. La latenza è la latenza dello shard più lento più il costo di merge. Il throughput viene diviso per N, perché ogni query ora consuma uno slot su ogni shard invece che su uno solo. Se fai troppe query in fan-out, il cluster gira di fatto a 1/N della sua capacità nominale. Ecco perché il fan-out deve essere l’eccezione, non la regola.
Quando i risultati parziali sono grandi (un sort globale su milioni di righe), lo step di merge diventa di per sé costoso. A volte il merge viene scaricato su una macchina separata. A volte l’applicazione accetta un’approssimazione top-K invece di un vero sort globale. A volte la query viene riscritta per usare uno store completamente diverso, che è la strada verso l’analytics warehouse trattata nella lezione 65.
flowchart LR
Q1[Query: orders for user 42] --> R1[Router]
R1 -->|hash user_id 42| S1[(Shard 7)]
S1 --> Resp1[1ms response]
Q2[Query: top 100 orders today] --> R2[Router]
R2 --> SA[(Shard 1)]
R2 --> SB[(Shard 2)]
R2 --> SC[(Shard ...)]
R2 --> SD[(Shard N)]
SA --> M[Merge top-K]
SB --> M
SC --> M
SD --> M
M --> Resp2[N-shard latency + merge]I due pattern di query messi fianco a fianco. La query a uno shard è un singolo hop. La query in fan-out è N hop in parallelo seguiti da un merge. La differenza di costo è ciò che rende la scelta della shard key la decisione più consequenziale del sistema.
Perché “shard by user_id” è diventata la religione
Tre ragioni, tutte da dire ad alta voce perché la scelta è così riflessa nel SaaS moderno che gli ingegneri spesso dimenticano che ci fosse una decisione da prendere.
I workload user-centric hanno query user-scoped. La grande maggioranza delle query in un’applicazione SaaS inizia con “per questo utente” o “in questo tenant”. Quando i dati sono disposti in modo che i dati di ogni utente vivano su uno shard, quelle query colpiscono uno shard. L’architettura si compone con il workload.
Le entità utente sono indipendenti. I miei dati non devono essere joinati ai tuoi per nessuna feature normale. Gli ordini, il profilo e la cronologia di un utente sono autonomi. Non c’è un’integrità relazionale intrinseca che attraversi il confine dell’utente, quindi la perdita dei join cross-user non è dolorosa.
Il reporting si può spostare offline. Le query che hanno davvero bisogno di una vista globale (dashboard di analytics, business intelligence, fraud detection tra utenti) non sono critiche per la latenza. Possono essere eseguite contro uno store di analytics separato che ingerisce dati da tutti gli shard via change-data-capture. Il cluster shardato transazionale viene lasciato tranquillo a fare il suo lavoro di query a uno shard. La lezione 65 copre questo pattern in dettaglio.
La combinazione fa sì che, per un SaaS user-centric, lo sharding per user_id sia quasi gratis. Quasi ogni query è naturalmente su uno shard, le poche che non lo sono possono essere spostate su un analytics warehouse, e il cluster scala con il numero di utenti senza sorprese architetturali.
Quando lo sharding per user_id non funziona
Tre situazioni, tutte da riconoscere.
Dati multi-tenant con viste cross-user dentro un tenant. Un SaaS B2B in cui un tenant ha molti utenti, e un admin di quel tenant vuole vedere tutta l’attività di tutti i suoi utenti, sarà scontento dello sharding per user_id. La query dell’admin è naturalmente circoscritta a un tenant, ma i dati sono sparsi su molti shard (uno per utente). Il fix è shardare per tenant_id invece. Ora una query dell’admin è uno shard, e una query utente è anch’essa uno shard (perché i dati dell’utente vivono sullo shard del tenant). Il trade è che un tenant molto grande diventa una hot partition, che è il problema coperto nella lezione 28.
Feature cross-user. Chat tra due utenti. Un grafo di amicizie. Un like sul post di qualcun altro. Queste feature attraversano intrinsecamente il confine dell’utente, e nessuna shard key le tiene su un solo shard. Le risposte pragmatiche: memorizzare i messaggi due volte, una sullo shard di ciascun partecipante, in modo che qualsiasi query su “i miei messaggi” rimanga sul mio shard. Oppure memorizzare i dati cross-user in uno store separato, più piccolo, non shardato, dedicato alla tabella delle relazioni. Oppure usare materialised view event-driven che pre-calcolano i join cross-user in inbox per utente. Tutte e tre sono comuni; nessuna è gratis.
Lookup globali per qualcosa che non sia user_id. “Trova l’utente con email alice@example.com” è una query globale quando si sharda per user_id, perché l’email vive su qualsiasi shard possieda quell’utente, e tu non sai quale senza un indice separato. Il fix standard è una piccola tabella di lookup globale (email -> user_id) mantenuta da ogni write su uno shard, o tenuta in uno store ausiliario non shardato. La tabella di lookup è abbastanza piccola da vivere su una macchina, il resto dei dati è shardato per scalare.
Il principio generale: quando individui una query che non si adatta alla shard key, hai tre opzioni oneste. Replicare i piccoli dati di riferimento su ogni shard in modo che ogni shard possa rispondere localmente. Costruire un indice secondario denormalizzato (spesso in uno store separato) in modo che anche il percorso di accesso alternativo sia una lookup a uno shard. Accettare il fan-out per quella specifica query e assicurarti che sia abbastanza rara da essere assorbita dal cluster.
Replicare le tabelle di riferimento
Un pattern che vale la pena di nominare. Alcune tabelle sono piccole, cambiano raramente, e vengono joinate da ogni query: codici nazione, tassi di cambio, definizioni dei feature flag, categorie di prodotto, limiti dei piani. Se shardi una tabella così, ogni join diventa un fan-out, il che rovina tutto.
Il fix è replicare la tabella su ogni shard. Ogni shard tiene una copia completa. I join sono locali. Gli update vengono scritti su ogni shard, il che è accettabile perché sono rari. I sistemi di sapore Postgres a volte hanno un supporto esplicito per questo: Citus le chiama “reference tables”, Vitess ha un concetto simile. In un’applicazione shardata fatta a mano, il sistema di migration spinge cambiamenti di schema e di dati a tutti gli shard.
Materialised view per la query scomoda
L’altra valvola di fuga. Quando una query davvero non può vivere su uno shard con il layout naturale dei dati, precalcoli la risposta in una struttura che invece può.
L’esempio classico è un feed. Il feed dell’utente A combina i post degli utenti B, C, D, E. Se i dati sono shardati per user_id, costruire il feed di A richiede un fan-out su tanti utenti quanti A ne segue. Il fix è mantenere il feed di A come una sua tabella, shardata per lo user_id di A. Quando B posta, un evento scrive una copia denormalizzata nelle tabelle del feed di chiunque segua B. Il costo di costruzione del feed viene pagato al momento della write, distribuito e ammortizzato. La read è una lookup a uno shard.
Questo è il pattern fan-out-on-write, ed è quello che ogni prodotto social di larga scala usa per i feed. Twitter ha parlato pubblicamente di fare esattamente questo. Il costo è reale: il post di B potrebbe essere scritto in un milione di tabelle del feed dei follower. La compensazione è una latenza di lettura costante indipendentemente da quanti utenti abbia B. Il Modulo 5 copre l’idraulica event-driven che rende tutto questo operativo.
I numeri di performance, in concreto
Una query a uno shard su Postgres contro un indice di primary key: da 0.5 a 2 ms, incluso il round-trip locale di rete. Il layer di routing aggiunge al massimo un millisecondo sopra.
Un fan-out su 10 shard sulla stessa rete: grossomodo da 5 a 10 ms, dominato dallo shard più lento più un millisecondo di merge. Accettabile occasionalmente, costoso a ogni page view. Un fan-out su 100 shard: da 10 a 30 ms nel caso migliore, molto peggio se anche un solo shard è lento.
Un fan-out cross-region, in cui gli shard vivono in region geografiche diverse: da 100 a 300 ms minimo, vincolato dalla latenza di rete inter-region. Questo è quasi sempre troppo lento per una query interattiva. La risposta standard è il geo-sharding per chiave di region e accettare che le query globali non siano interattive. Questi numeri sono la ragione per cui l’architettura è costruita attorno alle query a uno shard.
La forma di un’applicazione shardata in salute
Per chiudere: che aspetto ha lo sharding fatto bene?
La shard key combacia con il pattern di query dominante. Quasi ogni query sull’hot path filtra per la shard key esplicitamente, in modo che il layer di routing possa inoltrare direttamente allo shard di destinazione. Un piccolo numero di tabelle di riferimento è replicato su ogni shard per i join locali. Esiste un piccolo numero di query cross-shard; sono o abbastanza rare da poter fare fan-out, o spostate in materialised view che le riportano a read a uno shard, o spostate in uno store di analytics che ingerisce CDC da tutti gli shard.
Quando l’applicazione cresce, vengono aggiunti più shard. Le query sull’hot path continuano a essere a uno shard. La capacità del cluster cresce linearmente con il numero di shard. Questo è quello che la gente intende quando dice “lo sharding funziona”. È il risultato dell’azzeccare la shard key e di trattare le query cross-shard come l’eccezione rara che devono essere.
La lezione 32, l’ultima del Modulo 4, percorre in dettaglio la migrazione dello storage di Discord. È l’esempio da manuale di un sistema la cui architettura è stata costruita attorno a un pattern di query e una shard key, e le cui migrazioni tra tre database diversi hanno tutte preservato quella stessa forma.
Riferimenti e ulteriori letture
- “Designing Data-Intensive Applications” (Martin Kleppmann, O’Reilly, 2017), capitolo 6. Il riferimento standard per partizionamento, shard key, e query cross-partition.
- Documentazione di Citus, “Reference Tables”,
https://docs.citusdata.com/en/stable/develop/reference_tables.html(consultato 2026-05-01). Il trattamento operativo delle tabelle di riferimento replicate in un prodotto di sharding per Postgres. - Documentazione di Vitess,
https://vitess.io/docs/(consultato 2026-05-01). Sharding di MySQL con supporto esplicito per i pattern discussi in questa lezione. - Twitter Engineering, “The Infrastructure Behind Twitter: Scale”,
https://blog.twitter.com/engineering/en_us/topics/infrastructure/2017/the-infrastructure-behind-twitter-scale(consultato 2026-05-01). Discussione pubblica del fan-out-on-write per la consegna dei feed su larga scala.