Creare una pipeline ETL con Airflow

Benvenuto nel mondo delle pipeline ETL con Apache Airflow. In questo tutorial ci concentreremo sul recupero di dati di borsa con la Polygon API, sulla loro trasformazione e sul caricamento in un database SQLite per un accesso e una manipolazione più semplici. Iniziamo!

Che cos’è Apache Airflow e cos’è l’ETL con Airflow?

Apache Airflow è considerato uno standard del settore per l’orchestrazione dei dati e la gestione delle pipeline. È diventato popolare tra data scientist, machine learning engineer e professionisti dell’AI per la sua capacità di orchestrare workflow complessi, gestire le dipendenze tra task, riprovare i task falliti e fornire log dettagliati.

ETL con Airflow si riferisce all’uso di Apache Airflow per gestire i processi ETL. Ricapitolando, l’ETL è un tipo di integrazione dei dati che prevede l’estrazione dei dati da varie fonti, la loro trasformazione in un formato adatto all’analisi e il caricamento in una destinazione finale come un data warehouse. 

Configurare il nostro ambiente di sviluppo Airflow 

Prima di poter iniziare a costruire una pipeline ETL con Airflow, dobbiamo configurare il nostro ambiente di sviluppo. Per informazioni dettagliate su come configurarlo, dai un’occhiata al nostro tutorial Primi passi con Apache Airflow.

Dovremo anche installare l’Astro CLI. Consulta Astronomer, che mantiene l’Astro CLI e fornisce documentazione dettagliata, per le istruzioni.

Creare un progetto Airflow

Dopo aver configurato l’ambiente e installato l’Astro CLI, creiamo un progetto Airflow. Lo facciamo aprendo un terminale e creando una nuova directory nel percorso scelto.

 ~/Documents/data-engineering/ETL-pipeline/ 

Dalla root di questa directory eseguiamo il seguente comando per creare le risorse necessarie:

astro dev init

Il contenuto della directory sarà simile a quanto riportato sotto. L’output esatto può variare.

├── dags/
├── include/
├── plugins/
├── tests/
├── airflow_settings.yaml
├── Dockerfile
├── packages.txt
└── requirements.txt

Per avviare il progetto, esegui il seguente comando: 

astro dev start

Il tuo ambiente Airflow impiegherà circa un minuto per avviarsi. Una volta pronto, vai su localhost: 8080 nel browser e verrai accolto dalla UI di Airflow.

Ora sei pronto per iniziare a sviluppare la tua pipeline ETL con Airflow!

Progettare una pipeline ETL

È importante prendersi il tempo per pianificare ogni componente della pipeline prima ancora di scrivere una riga di codice. In particolare, è una buona abitudine identificare prima la fonte dei dati e la destinazione in cui verranno caricati. Dedicando un po’ di tempo a individuare le fonti e la destinazione, capiremo anche come i dati verranno trasformati lungo il percorso.

Nel nostro esempio, progetteremo una pipeline di dati per prelevare dati di borsa dalla Polygon API, quindi trasformarli e caricarli in un database SQLite. In questo caso, il sistema sorgente è la Polygon API e la destinazione è un database SQLite. Illustriamo il tutto con un’immagine: 

Diagramma di sorgente e destinazione

Dalla nostra esperienza come data engineer sappiamo che, per preparare i dati al caricamento in un database SQLite, è necessario trasformarli da JSON a un formato tabellare. Creiamo un piano per trasformare i dati dopo averli prelevati dalla Polygon API, usando Python nativo e la libreria pandas. Aggiorniamo il nostro schema per mostrare questa modifica:

Diagramma architetturale

Aggiungendo queste informazioni aggiuntive, abbiamo creato un diagramma architetturale, una rappresentazione visiva generale del nostro sistema. Vediamo i tre passaggi logici di questa pipeline, che corrispondono alla E, T e L del nostro processo.

Possiamo anche tradurre ciascuno di questi task in un grafo aciclico orientato, o DAG, ovvero una configurazione specifica che definisce l’insieme completo dei task eseguiti da Airflow, la loro sequenza e le dipendenze reciproche. Scopri di più sui grafi aciclici orientati con il nostro corso Introduzione alla Data Engineering, che approfondisce i DAG di Airflow.

In un contesto enterprise, molti team usano altri tipi di documenti chiamati specifiche tecniche, o tech spec, per concordare e documentare le scelte di design. Nel nostro caso, è sufficiente creare una tabella per documentare le scelte progettuali della pipeline di dati. 

Scelte per la pipeline di dati

Tieni presente che il nostro DAG è focalizzato sui dettagli a livello di task e potrebbe non coprire ogni informazione. Possiamo usare una seconda tabella per documentare ulteriori dettagli da definire. Le nostre domande includono:

• Con quale frequenza verrà eseguito questo DAG? 
• Cosa succede se un task nella pipeline fallisce? 
• E se volessimo raccogliere dati su ulteriori titoli?

Questioni da definire

Facciamo il punto su ciò che abbiamo fatto: abbiamo creato un diagramma architetturale, documentato come scomporre la pipeline in task di Airflow e individuato le informazioni di alto livello necessarie per configurare il DAG.

Se vuoi saperne di più su progettazione, sviluppo e test delle pipeline di dati, dai un’occhiata al corso di DataCamp Introduzione alle data pipeline. Imparerai le basi per costruire pipeline ETL con Python, oltre alle best practice per rendere la tua soluzione robusta, resiliente e riutilizzabile.

Creare una pipeline ETL con Airflow

Organizzeremo lo sviluppo della pipeline ETL seguendo i passaggi in ordine. Un approccio strutturato assicura che ogni fase venga eseguita con precisione. 

Estrazione dei dati con Airflow

Prima di prelevare i dati dalla Polygon API, dobbiamo creare un token API visitando Polygon e selezionando il pulsante Create API Key. Nota che per questo tutorial non serve sottoscrivere un abbonamento a pagamento: il piano gratuito genera una chiave API e offre tutte le funzionalità necessarie. Ricorda solo di copiare e salvare la tua chiave API.

Dopo aver creato una chiave API, siamo pronti a iniziare a estrarre i dati dalla Polygon API con Airflow. Usiamo la tabella delle specifiche tecniche che abbiamo creato, che include i dettagli per la configurazione del DAG, per aiutarci a scrivere il codice. 

from airflow import DAG 
from datetime import datetime, timedelta

with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=True,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:

Aggiungiamo il primo task. Useremo la TaskFlow API e il modulo requests per recuperare i dati di apertura/chiusura dei titoli dalla Polygon API. 

import requests
...
@task()

def hit_polygon_api(**context):
    # Instantiate a list of tickers that will be pulled and looped over
    stock_ticker = "AMZN"
    # Set variables
    polygon_api_key= "<your-api-key>"
  ds = context.get("ds")

  # Create the URL
    url= f"<https://api.polygon.io/v1/open-close/{stock_ticker}/{ds}?adjusted=true&apiKey={polygon_api_key}>"
    response = requests.get(url)
    # Return the raw data
    return response.json()

Ci sono alcune cose da notare in questo codice. 

• La funzione hit_polygon_api è decorata con @task. Questo decorator converte la funzione in un task Airflow eseguibile come parte di un DAG.
• Il parametro context è definito nella signature di hit_polygon_api. Viene poi usato per estrarre il valore associato alla chiave ds.
• context è un dizionario che contiene metadati sul task e sul DAG.
• Quando recuperiamo ds dal dizionario context, otteniamo la data di data_interaval_end nel formato YYYY-mm-dd.  
• Per assicurarci che il nuovo task venga eseguito quando gira il DAG, dobbiamo aggiungere una chiamata a hit_polygon_api.

Unendo tutto, il codice per creare la prima parte della nostra pipeline ETL è il seguente.

from airflow import DAG
from airflow.decorators import task
from datetime import datetime, timedelta import requests

with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=False,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:
    # Create a task using the TaskFlow API
    @task()
    def hit_polygon_api(**context):
        # Instantiate a list of tickers that will be pulled and looped over
        stock_ticker = "AMZN"
        # Set variables
        polygon_api_key = "<your-api-key>"
        ds = context.get("ds")
        # Create the URL
        url = f"<https://api.polygon.io/v1/open-close/{stock_ticker}/{ds}?adjusted=true&apiKey={polygon_api_key}>"
        response = requests.get(url)
        # Return the raw data
        return response.json()
    hit_polygon_api()

Noterai che la risposta conterrà un errore quando viene eseguita per il 1° gennaio 2024. Poiché il mercato è chiuso in quel giorno, Polygon restituisce una risposta che segnala l’eccezione. Ce ne occuperemo nel prossimo passaggio.

Trasformazione dei dati con Airflow

Dopo aver estratto i dati dalla Polygon API, siamo pronti a trasformarli.

Per farlo, creeremo un altro task usando la TaskFlow API. Questo task si chiamerà flatten_market_data e avrà i parametri polygon_response, ossia i dati grezzi restituiti dalla funzione hit_polygon_api, e **context. Tra poco daremo uno sguardo più da vicino a polygon_response.

La trasformazione che effettueremo è piuttosto semplice. Appiattiremo il JSON restituito dalla Polygon API in una lista. La particolarità è che forniremo valori predefiniti univoci per ogni chiave se non esiste nella risposta.

Ad esempio, se la chiave from non è presente nella risposta, forniremo un valore predefinito usando il context di Airflow. Questo risolve il problema visto in precedenza, quando una risposta conteneva un set limitato di chiavi (poiché il mercato era chiuso). Successivamente, convertirermo la lista in un DataFrame di pandas e lo restituiremo. Il task risultante è il seguente:

@task
def flatten_market_data(polygon_response, **context):
    # Create a list of headers and a list to store the normalized data in
    columns = {
        "status": "closed",
        "from": context.get("ds"),
        "symbol": "AMZN",
        "open": None,
        "high": None,
        "low": None,
        "close": None,
        "volume": None
    }
    # Create a list to append the data to
    flattened_record = []
    for header_name, default_value in columns.items():
        # Append the data
        flattened_record.append(polygon_response.get(header_name, default_value))
    # Convert to a pandas DataFrame
    flattened_dataframe = pd.DataFrame([flattened_record], columns=columns.keys())
    return flattened_dataframe

Dobbiamo aggiungere una dipendenza tra i task hit_polygon_api e flatten_market_data. Per farlo, aggiorniamo il codice del nostro DAG come qui sotto:

import pandas as pd
with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=True,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:
    @task()
    def hit_polygon_api(**context):
        ...
    @task
    def flatten_market_data(polygon_response, **context):
        # Create a list of headers and a list to store the normalized data in
        columns = {
            "status": None,
            "from": context.get("ds"),
            "symbol": "AMZN",
            "open": None,
            "high": None,
            "low": None,
            "close": None,
            "volume": None
        }
        # Create a list to append the data to
        flattened_record = []
        for header_name, default_value in columns.items():
            # Append the data
            flattened_record.append(polygon_response.get(header_name, default_value))
        # Convert to a pandas DataFrame
        flattened_dataframe = pd.DataFrame([flattened_record], columns=columns.keys())
        return flattened_dataframe
# Set dependencies
    raw_market_data = hit_polygon_api()
    flatten_market_data(raw_market_data)

Qui, il valore restituito dal task hit_polygon_api viene memorizzato in raw_market_data. Poi, raw_market_data viene passato come argomento al task flatten_market_data tramite il parametro polygon_response. Questo codice non solo imposta una dipendenza tra i task hit_polygon_api e flatten_market_data, ma consente anche la condivisione dei dati tra i due task.

Sebbene la trasformazione eseguita sia stata relativamente semplice, Airflow offre la possibilità di eseguire un’ampia gamma di manipolazioni dei dati più complesse. Oltre ai task nativi, è facile sfruttare l’ampia raccolta di hook e operator dei provider di Airflow per orchestrare trasformazioni con strumenti come AWS Lambda e DBT.

Caricamento dei dati con Airflow

Siamo arrivati all’ultimo passaggio della nostra pipeline ETL. Abbiamo pianificato di farlo usando un database SQLite e un ultimo task definito con la TaskFlow API.

Come prima, definiremo un solo parametro quando creiamo il task. Lo chiameremo flattened_dataframe. In questo modo i dati restituiti dal task flatten_market_data possono essere passati al nuovo task.

Prima di scrivere il codice per caricare i dati in un database SQLite, dobbiamo prima creare una connessione nella UI di Airflow. Per aprire la pagina delle connessioni, segui questi passaggi:

• Apri la UI di Airflow
• Passa con il mouse su Admin
• Seleziona Connections. 
• Clicca sull’icona + per creare una nuova connessione.

Verrai reindirizzato a una schermata come questa:

Pagina delle connessioni di Airflow

Per compilare la pagina delle connessioni, segui questi passaggi:

• Imposta Connection Type su Sqlite.
• Inserisci il valore "market_database_conn" per Connection Id.
• Aggiungi "/usr/local/airflow/market_data.db" nel campo Host.

La configurazione di questa connessione dovrebbe essere come nell’immagine seguente. Una volta fatto, clicca su Save.

Connessione Airflow a SQLite non ancora salvata

Ora che abbiamo creato una connessione, possiamo recuperare queste informazioni nel nostro task usando la SqliteHook. Guarda il codice qui sotto.

from airflow.providers.sqlite.hooks.sqlite import SqliteHook

@task
def load_market_data(flattened_dataframe):
    # Pull the connection
    market_database_hook = SqliteHook("market_database_conn")
    market_database_conn = market_database_hook.get_sqlalchemy_engine()
    # Load the table to Postgres, replace if it exists
    flattened_dataframe.to_sql(
        name="market_data",
        con=market_database_conn,
        if_exists="append",
        index=False
    )
    # print(market_database_hook.get_records("SELECT * FROM market_data;"))

Con questo codice creiamo una connessione al database SQLite specificato nel passaggio precedente. Quindi, il motore di connessione viene recuperato dall’hook tramite il metodo .get_sqlalchemy_engine(). Questo viene poi passato come argomento al parametro con quando il metodo .to_sql() viene chiamato su flattened_dataframe.

Nota che il nome della tabella su cui vengono scritti i dati è market_data e, se la tabella esiste, i dati vengono aggiunti in append. Durante i test, mi piace verificare che i dati vengano scritti recuperando e stampando questi record. Puoi farlo decommentando l’ultima riga di questo task.

Mettendo tutto insieme, il nostro codice dovrebbe apparire più o meno così:

from airflow import DAG
from airflow.decorators import task
from airflow.providers.sqlite.hooks.sqlite import SqliteHook
from datetime import datetime, timedelta
import requests
import pandas as pd

with DAG(
    dag_id="market_etl",
    start_date=datetime(2024, 1, 1, 9),
    schedule="@daily",
    catchup=True,
    max_active_runs=1,
    default_args={
        "retries": 3,
        "retry_delay": timedelta(minutes=5)
    }
) as dag:
    # Create a task using the TaskFlow API
    @task()
    def hit_polygon_api(**context):
        # Instantiate a list of tickers that will be pulled and looped over
        stock_ticker = "AMZN"
        # Set variables
        polygon_api_key = "<your-api-key>"
        ds = context.get("ds")
        # Create the URL
        url = f"<https://api.polygon.io/v1/open-close/{stock_ticker}/{ds}?adjusted=true&apiKey={polygon_api_key}>"
        response = requests.get(url)
        # Return the raw data
        return response.json()
    @task
    def flatten_market_data(polygon_response, **context):
        # Create a list of headers and a list to store the normalized data in
        columns = {
            "status": None,
            "from": context.get("ds"),
            "symbol": "AMZN",
            "open": None,
            "high": None,
            "low": None,
            "close": None,
            "volume": None
        }
        # Create a list to append the data to
        flattened_record = []
        for header_name, default_value in columns.items():
            # Append the data
            flattened_record.append(polygon_response.get(header_name, default_value))
        # Convert to a pandas DataFrame
        flattened_dataframe = pd.DataFrame([flattened_record], columns=columns.keys())
        return flattened_dataframe
    @task
    def load_market_data(flattened_dataframe):
        # Pull the connection
        market_database_hook = SqliteHook("market_database_conn")
        market_database_conn = market_database_hook.get_sqlalchemy_engine()
        # Load the table to SQLite, append if it exists
        flattened_dataframe.to_sql(
            name="market_data",
            con=market_database_conn,
            if_exists="append",
            index=False
        )
    # Set dependencies between tasks
    raw_market_data = hit_polygon_api()
    transformed_market_data = flatten_market_data(raw_market_data)
    load_market_data(transformed_market_data)

Anche qui, abbiamo aggiornato le dipendenze per consentire ai dati restituiti dal task flatten_market_data di essere passati al task load_market_data. La vista grafica risultante del nostro DAG è più o meno questa:

Vista a grafo per la pipeline ETL

Test

Ora che hai creato il tuo primo DAG in Airflow, è il momento di assicurarti che funzioni. Ci sono vari modi per farlo, ma uno dei più comuni è eseguire il DAG end-to-end.

Per farlo nella UI di Airflow, vai al tuo DAG e attiva l’interruttore da blu a active. Poiché catchup è impostato su True, un’esecuzione del DAG verrà messa in coda e inizierà a girare. Se un task viene eseguito correttamente, il riquadro corrispondente nella UI diventerà verde. Se tutti i task del DAG hanno successo, il DAG sarà contrassegnato come success e verrà avviata l’esecuzione successiva.

Se un task fallisce, lo stato sarà up for retry e verrà segnato in giallo. In questo caso, è meglio dare un’occhiata ai log di quel task. Puoi farlo cliccando sul riquadro giallo nella vista a griglia e selezionando Logs. Qui troverai il messaggio di eccezione e potrai iniziare a fare triage. Se un task fallisce più volte del numero di retry specificato, lo stato di quel task e del DAG verrà impostato su failed.

Oltre a testare un DAG end-to-end, Airflow semplifica la scrittura di unit test. Quando crei inizialmente l’ambiente usando astro dev start, viene creata per te una directory tests/. Qui puoi aggiungere unit test sia per il tuo DAG sia per i componenti del DAG.

Qui sotto trovi un unit test per la configurazione del nostro DAG. Questo test convalida diversi parametri impostati in fase di definizione, come start_date, schedule e catchup. Una volta scritto il test, vai alla root del progetto ed esegui:

from airflow.models.dagbag import DagBag
from datetime import datetime
import pytz

def test_market_etl_config():
    # Pull the DAG
    market_etl_dag = DagBag().get_dag("market_etl")
    # Assert start date, schedule, and catchup
    assert market_etl_dag.start_date == datetime(2024, 3, 25, 9, tzinfo=pytz.UTC)
    assert market_etl_dag.schedule_interval == "@daily"
    assert market_etl_dag.catchup

astro dev pytest

Questo comando eseguirà tutti gli unit test nella directory tests/. Se vuoi eseguire un singolo test, puoi fornire il percorso al file come argomento al comando sopra. Oltre a usare l’Astro CLI per eseguire i test, è possibile utilizzare qualsiasi test runner Python per scrivere ed eseguire unit test.

Per progetti personali, scrivere unit test aiuta a garantire che il codice si comporti come previsto. In ambito enterprise, gli unit test sono quasi sempre richiesti. La maggior parte dei team di data engineering utilizzerà qualche strumento di CI/CD per distribuire il progetto Airflow. Questo processo in genere prevede l’esecuzione degli unit test e la validazione del loro output per assicurarsi che il DAG che hai scritto sia pronto per la produzione. Per ulteriori informazioni sugli unit test, consulta il nostro tutorial Come usare Pytest per i test unitari e il corso Introduzione al testing in Python.

Suggerimenti e tecniche avanzate per Airflow

Abbiamo creato una semplice pipeline di dati che funziona, e abbiamo persino usato tecniche di trasformazione e persistenza. In altri casi, Airflow è in grado di orchestrare workflow complessi usando operator forniti dai provider e operator personalizzati, elaborando terabyte di dati.

Alcuni esempi sono S3ToSnowflakeOperator e DatabricksRunNowOperator, che consentono facilmente l’integrazione con uno stack dati più ampio. Lavorare con questi operator è complicato in un contesto hobbistico. Ad esempio, per usare S3ToSnowflakeOperator servirebbero sia account AWS che Snowflake e la configurazione delle risorse tra cui trasferire i dati.

Oltre ai workflow ETL, Airflow supporta i workflow ELT, che stanno rapidamente diventando lo standard del settore per i team che utilizzano data warehouse in cloud. Tienilo a mente quando progetti le pipeline di dati.

Nella componente di load della nostra pipeline, abbiamo creato una connessione a un database SQLite, che è stata poi recuperata e utilizzata per persistere i dati. Le connessioni, talvolta chiamate Secrets, sono una funzionalità di Airflow pensata per semplificare le interazioni con i sistemi sorgente e destinazione. Usando queste connessioni per archiviare informazioni sensibili, come la tua chiave API di Polygon, migliori la sicurezza del codice. Questo approccio consente anche di gestire le credenziali separatamente dal codebase. Quando possibile, è consigliabile usare ampiamente le Connections per mantenere il workflow sia sicuro che ordinato.

Avrai notato che il ticker “AMZN” è stato hardcodato nei task hit_polygon_api e flatten_market_data. Questo ci ha permesso di estrarre, trasformare e caricare i dati per un singolo ticker. Ma se volessi usare lo stesso codice per più ticker? Fortunatamente è facile generare DAG dinamicamente. Con un refactoring minimo del codice, potremmo ciclare su una lista di ticker e parametrizzare i valori del ticker. Questo rende i DAG più modulari e portabili. Per informazioni sulla generazione dinamica dei DAG, consulta la documentazione di Astronomer Dynamically Generate DAGs in Airflow.

Conclusione

Congratulazioni! Hai creato un DAG di Airflow per estrarre, trasformare e caricare dati di borsa dalla Polygon API usando Python, pandas e SQLite. Lungo il percorso, hai affinato le tue abilità nel costruire diagrammi architetturali e tech spec, creare connessioni in Airflow e testare i DAG. Continuando il tuo percorso con Airflow, sperimenta tecniche più avanzate per rendere le tue pipeline robuste, resilienti e riutilizzabili.  

Dai un’occhiata a queste risorse per saperne di più: 

• Corso Introduction to Airflow in Python
• Tutorial Primi passi con Apache Airflow
• Confronto Airflow vs Prefect
• Le migliori alternative ad Airflow
• Le principali domande di colloquio su Airflow