Tag: etl, elt, data-pipeline, airflow, dbt, sql, data-engineering

Livello: intermedio

TL;DR: Pattern architetturali, differenze tra ETL ed ELT, strategie di caricamento incrementale, idempotenza e gestione degli errori nelle pipeline dati.

Concetti chiave

ETL estrae dati da sorgenti eterogenee, applica trasformazioni su un server di staging e carica il risultato nel target.

ELT carica i dati grezzi direttamente nel data warehouse, sfruttando la potenza di calcolo della destinazione per trasformarli.

La fase di Extract gestisce connessioni via API REST, query SQL, CDC (Change Data Capture) o dump di file.

Change Data Capture traccia gli aggiornamenti direttamente dal transaction log del database sorgente.

La fase di Transform include data cleansing, deduplicazione, normalizzazione, masking dei dati sensibili e aggregazioni di business.

La fase di Load scrive i dati trasformati in target relazionali, columnar warehouse o data lake.

Il caricamento Full Refresh sovrascrive interamente la destinazione ad ogni esecuzione.

Il caricamento Incremental applica filtri basati su timestamp o watermarking per processare solo i dati modificati.

L'idempotenza garantisce che eseguire la pipeline piu volte sullo stesso input produca sempre lo stesso output senza duplicati.

L'orchestrazione gestisce dipendenze, retry policy, backfill, logging e monitoraggio tramite DAG schedulati.

Domande tipiche e risposte pronte

Q: Qual e la differenza chiave tra una pipeline ETL e una pipeline ELT?R: Con ETL trasformo i dati fuori dal data warehouse prima di caricarli, mentre con ELT carico i dati grezzi direttamente nel warehouse e demando le trasformazioni a query SQL analitiche via dbt o motori columnar.

Q: Come gestisci l'idempotenza durante la fase di caricamento?R: Uso pattern di upsert basati su chiavi univoche oppure carico i dati in una staging table e applico una transazione atomica con MERGE INTO o DELETE mirata della partizione seguita da INSERT.

Q: Come implementi il caricamento incrementale senza Change Data Capture nativo?R: Salvo l'ultimo timestamp o watermarking dell'esecuzione precedente e interrogo la sorgente con un filtro strettamente superiore a tale valore, gestendo eventuali record sovrapposti tramite deduplicazione.

Q: Come gestisci i record malformati o non conformi allo schema durante una pipeline?R: Isolo i record non validi indirizzandoli verso una Dead Letter Queue o tabella di reject, registrando l'errore senza bloccare l'elaborazione del resto del batch.

Insidie e errori comuni

Mancata gestione del drift dello schema sorgente che causa fallimenti improvvisi nelle query di estrazione o caricamento.

Uso di INSERT INTO senza clausola di conflitto, con conseguente duplicazione dei record in caso di retry automatico del job.

Trasformazioni eseguite interamente in memoria che generano errori di Out Of Memory su volumi di dati imprevisti.

Mancanza di gestione del late-arriving data, che porta alla perdita di transazioni retrodatate nei filtri a finestra temporale rigida.

Assenza di transazionalita nel caricamento, lasciando la destinazione in uno stato inconsistente o parzialmente popolato se il job fallisce.

Snippet e comandiSQL

```-- Pattern idempotente MERGE INTO per aggiornamento incrementale

MERGE INTO target_schema.customers AS target

USING staging_schema.stg_customers AS source

ON target.customer_id = source.customer_id

WHEN MATCHED AND target.updated_at < source.updated_at THEN

UPDATE SET

target.name = source.name,

target.email = source.email,

target.updated_at = source.updated_at

WHEN NOT MATCHED THEN

INSERT (customer_id, name, email, updated_at)

VALUES (source.customer_id, source.name, source.email, source.updated_at);

Esempio Python di estrazione a batch con watermarking

def extract_batch(cursor, last_watermark, batch_size=5000):

query = """

SELECT id, payload, updated_at

FROM events

WHERE updated_at > %s

ORDER BY updated_at ASC

LIMIT %s

"""

cursor.execute(query, (last_watermark, batch_size))

return cursor.fetchall()

```

Da ripassare

[ ] Pattern di Change Data Capture tramite Debezium e log parsing

[ ] Strategie di partition switching e stage table swap nei data warehouse columnar

[x] Differenze architetturali tra batch processing e streaming continuo

[x] Utilizzo dell'istruzione SQL MERGE per la deduplicazione e l'idempotenza