Foto di eMotion Tech su Unsplash
Premessa
Questo articolo presenta un approccio per affrontare una sfida ben nota nei sistemi di data warehouse: mantenere i dati freschi e aggiornati senza ricorrere a mutazioni su larga scala.
In DoiT affianchiamo molti clienti nella progettazione di sistemi well-architected e nell'uso efficiente dei servizi cloud; quanto segue nasce proprio dall'esperienza con un cliente.
Contesto
Un'azienda SaaS offre ai propri clienti una piattaforma analitica costruita su BigQuery. Per alcuni di loro, i dati arrivano sotto forma di snapshot completo, comprensivo dell'intero storico e non solo delle variazioni incrementali rispetto all'ultimo aggiornamento.
Uno di questi clienti fornisce uno snapshot da 15TB, mentre i dati effettivamente aggiornati rappresentano appena lo 0,1%. Non potendo ricevere solo gli aggiornamenti incrementali, il team ha dovuto individuare il metodo più affidabile ed efficiente — sia in termini di prestazioni sia di costi — per mantenere aggiornata la tabella esistente con i dati dei nuovi snapshot.
Requisiti della data pipeline
L'azienda e il cliente hanno definito un data contract per la pipeline, in modo da soddisfare i requisiti di business e tecnici:
- I dati in ingresso sono uno snapshot completo su una finestra mobile di due anni, con cadenza di sei volte al giorno.
- I dati sono partizionati per giorno (ogni partizione pesa circa 20GiB) e contrassegnati da uno snapshot ID (valore incrementale).
- I dati comprendono sia record nuovi (nuove chiavi) sia record modificati (aggiornamenti di dati esistenti).
- Scadenza delle partizioni configurata a 2 anni.
- I record presenti nello snapshot corrente ma assenti nel nuovo snapshot restano fino alla scadenza della partizione.
Il team prodotto dell'azienda SaaS ha definito i seguenti requisiti:
- Modellazione: due tabelle conterranno rispettivamente i dati esistenti e quelli nuovi.
I nuovi dati risiedono in una tabella chiamata "staging", mentre i dati esistenti (quelli esposti al cliente) si trovano in una tabella chiamata "target".
- Univocità: i dati della tabella "target" sono univoci (nessuna chiave duplicata).
- Freschezza: la tabella "target" si aggiorna entro un'ora dall'arrivo dello snapshot.
- Disponibilità: l'utente finale riceve sempre una risposta alle proprie query.
La sfida delle mutazioni
Come spiega il blog di Google citato qui sotto, BigQuery non è progettato come i database OLTP transazionali, nativamente efficienti nel gestire mutazioni di grande entità.
"BigQuery is not unique among OLAP databases in being constrained on mutation frequency, either explicitly (through quotas) or implicitly (resulting in significant performance degradation), since these types of databases are optimized for large-scale ingestion and analytical queries as opposed to transaction processing. Furthermore, BigQuery allows you to read the state of a table at any instant in time during the preceding seven days. This look-back window requires retaining data that the user has already deleted. To support cost-effective and efficient queries at scale, BigQuery limits the frequency of mutations using quotas." ( Performing large-scale mutations in BigQuery | Google Cloud Blog )
Una sostituzione completa che rimuove il vecchio snapshot e lo rimpiazza con quello nuovo non basta, perché occorre conservare i record presenti nello snapshot corrente ma assenti in quello più recente.
Serve quindi confrontare i record tra lo snapshot esistente e quello nuovo, mentre il classico approccio di sostituzione non effettua alcun confronto.
Per illustrare le opzioni di mutazione e le relative differenze, definiamo il problema in modo puntuale, con la modellazione dei dati e la logica di mutazione esatte.
Modellazione dei dati
Schema della tabella:
Un record nella tabella ("staging" o "target") contiene un attribute_id e un valore per uno specifico prodotto (product_id) in un negozio (store_id), per una data specifica, oltre al riferimento allo snapshot in cui era presente (update_id).
Esempio di record:
In base ai dati ricevuti nello snapshot ID 1 (update_id), il valore della quantità totale (attribute_id = 1) di un prodotto (product_id = 301865) in un negozio (store_id = 2072) era pari a 20 in data 2017–04–21.
Esempio di logica di mutazione:
Quello che segue è un campione tratto da una partizione (data specifica) appartenente a due snapshot diversi.
Tabella "target" dopo l'ingestione dello snapshot #1
Lo snapshot #2 conteneva le seguenti modifiche:
1. Blu: record aggiornati (presenti in entrambi gli snapshot), con valore modificato.
2. Rosso: record storico (presente solo nello snapshot #1).
3. Verde — nuovo record (presente solo nello snapshot #2).
Tabella "target" dopo l'ingestione dello snapshot #2
Poiché BigQuery supporta MERGE, l'approccio più immediato è unire la tabella di staging con quella target seguendo questa logica.
Logica di MERGE
Questa logica di MERGE definisce le mutazioni da applicare ai diversi casi, in base a una chiave di join formata dai seguenti campi: product_id, store_id, attribute_id.
Questa è l'implementazione in codice della logica di MERGE:
MERGEnw-playground.demo.merge_target TUSINGnw-playground.demo.merge_staging SON T.product_Id = S.product_Id AND T.store_Id = S.store_Id AND T.attribute_Id = S.attribute_Id AND T.date = S.date WHEN MATCHED and T.date >= '2017-04-14' and T.date <= '2017-04-26' THEN UPDATE SET T.attr_value = S.attr_value, T.update_id = S.update_id WHEN NOT MATCHED BY TARGET and S.date >= '2017-04-14' and S.date <= '2017-04-26' THENINSERT (date, product_Id, store_Id, attribute_Id, attr_value, update_id)VALUES (S.date, S.Product_id, S.store_Id, S.attribute_Id, S.attr_value, S.update_id);Il problema:
Come anticipato, l'operazione di MERGE in BQ è molto onerosa e presenta limitazioni, perché richiede shuffling e mutazioni dei dati esistenti.
Dobbiamo quindi trovare un modo per evitare il MERGE e la mutazione dei dati.
Approccio basato su deduplica e CLONE
Cosa faremo?
Invece di unire i record delle tabelle "staging" e "target", procederemo nel modo seguente:
- Aggiungiamo (append) il nuovo snapshot alla tabella "staging".
- Deduplichiamo i record simili mantenendo il più recente.
La deduplica è il processo che elimina i record duplicati lasciandone uno solo. Si ottiene definendo una chiave per identificare i record simili e una logica per decidere quale conservare tra i duplicati (ad esempio, il più recente). 3. Sostituiamo la tabella "staging" con i risultati della deduplica. 4. Sostituiamo la tabella "target" con un CLONE della tabella "staging".
"CLONE A table clone is a lightweight, writeable copy of another table (called the base table). You are only charged for data storage in the table clone that differs from the base table, so initially, there is no storage cost for a table clone. Other than the billing model for storage and some additional metadata for the base table, a table clone is similar to a standard table — you can query it, make a copy of it, delete it, and so on. After you create a table clone, it is independent of the base table. Any changes made to the base table or table clone aren't reflected in the other." table-clones-intro
Come lo faremo?
- Append dei dati: tramite il comando
LOADin modalità "append", così che la tabella di staging contenga sia lo snapshot corrente sia quello nuovo. - Deduplica: dedupliamo i record simili (stessa chiave) mantenendo il più recente. La deduplica non tocca i record completamente nuovi o presenti solo nello snapshot corrente. È implementata con le window functions, che calcolano valori su un gruppo di righe, e con l'operazione
QUALIFY. La clausolaQUALIFYconsente di filtrare i risultati di una window function (gruppo di righe). - Sostituzione di "staging": tramite
CREATE OR REPLACEsui risultati della query di deduplica. - Clonazione di "staging" su "target": tramite
CREATE OR REPLACE TABLE ... CLONE.
Perché lo facciamo?
Rispetto all'approccio con MERGE, questa strategia offre diversi vantaggi:
1. Meno shuffling dei dati: aggiungere il nuovo snapshot a quello esistente evita lo shuffling tra due tabelle.
2. Niente operazioni MERGE/JOIN: la deduplica si realizza con una window function che usa la clausola PARTITION BY, la quale suddivide le righe in input in partizioni separate, su cui la window function viene valutata in modo indipendente.
3. Niente mutazioni: si creano o clonano nuove tabelle, evitando i limiti delle quote di mutazione di BigQuery.
Esempio di logica di append e deduplica:
Tabella di staging dopo l'append del nuovo snapshot (#2)
Analogamente al risultato del MERGE, la tabella dopo la deduplica contiene le seguenti modifiche:
1. Blu — record aggiornati (presenti in entrambi gli snapshot), con valore modificato.
2. Rosso — record storici da conservare (presenti solo nello snapshot #1).
3. Verde — nuovo record (presente solo nello snapshot #2).
Tabella di staging dopo la deduplica
All'arrivo dello snapshot successivo, il processo si ripete con le stesse modalità.
Esempi di codice
Il codice seguente crea una nuova tabella di staging a partire dai risultati della deduplica, sostituendo la tabella esistente. Sfrutta la clausola QUALIFY e assegna un row number a ogni record con chiave simile: il primo record (rownum=1) è il più recente.
CREATE OR REPLACE TABLE `nw-playground.demo.dedup_staging` (date DATE, product_id INT64, store_id INT64, attribute_id INT64, attr_value INT64, update_id INT64)PARTITION BY dateCLUSTER BY product_id, store_id ASSELECT date, product_id, store_id, attribute_id, attr_value, update_idFROM `nw-playground.demo.dedup_staging`WHERE date >= '2017-04-14' AND date <= '2017-04-26' QUALIFY ROW_NUMBER() OVER(PARTITION BY date, product_id, store_id, attribute_id ORDER BY update_id DESC ) = 1;
CREATE OR REPLACE TABLE`nw-playground.nw_demo.dedup_target`CLONE `nw-playground.nw_demo.dedup_staging`;Dati di test e risultati
I test sono stati eseguiti su 7,8 miliardi di record (350GiB) distribuiti in 13 partizioni con circa 3M di differenze (di tutti i tipi) tra "staging" e "target", utilizzando il modello di fatturazione on-demand.
Nel caso del MERGE, la tabella "staging" conteneva solo il nuovo snapshot e la "target" lo snapshot esistente. Nel caso della deduplica, invece, la tabella di staging conteneva sia lo snapshot esistente sia quello nuovo, mentre la "target" era un clone del primo snapshot di staging.
Confronto dei risultati
La tabella seguente mette a confronto i risultati:
Riepilogo del confronto sui dettagli di esecuzione
Confrontando i piani di esecuzione, emerge chiaramente che la durata e l'utilizzo degli slot del test con MERGE sono il doppio rispetto a quelli del test con DEDUPLICATE:
Durata e utilizzo degli slot dell'operazione principale
Confronto dei costi
Modello di fatturazione on-demand
Il costo è simile tra i due test, perché la quantità di dati scansionati è identica. In un full load si arriva a circa 187$ (6,25$ per 1TiB), considerando 30TB e 2 snapshot.
BigQuery Editions
Il nuovo modello di fatturazione introdotto da Google nell'aprile 2023 si basa sul capacity pricing (pagamento per slot/ora) ed è disponibile in tre edizioni. Per stimare il costo del carico di produzione possiamo proiettare un full snapshot load applicando un fattore 56 rispetto al test eseguito (730 partizioni-giorno in 2 anni: 730/13 = 56).
La stima dei costi non è precisa, perché entrano in gioco diverse configurazioni e commitments. Serve soprattutto a evidenziare la differenza tra le due opzioni.
Confronto della stima dei costi tra le BigQuery Editions (prezzi US)
Appendice
Esistono diversi modi per implementare la deduplica (con ROWNUM o GROUP BY). La deduplica con ROWNUM ha mostrato prestazioni paragonabili a quelle della clausola QUALIFY, che però rimane una soluzione più pulita. L'uso di GROUP BY ha invece prodotto prestazioni peggiori, in linea con i risultati del MERGE.
Esiste anche un altro approccio efficiente: le stored procedure per Apache Spark, di cui parleremo nel prossimo articolo…
Come illustrato nel caso d'uso, si tratta di una mutazione su larga scala che si appoggia a una logica di join con i dati esistenti. La strada migliore per affrontarla è evitare le modifiche ai dati.
L'approccio basato su deduplica e CLONE permette di aggirare i limiti del DML e migliora sensibilmente entrambe le dimensioni — Elapsed Time e utilizzo degli slot — con una riduzione di prestazioni e costi pari a circa il 50%.