Framework per la pianificazione dei workflow

Introduzione al Data Engineering

Vincent Vankrunkelsven

Data Engineer, DataCamp

Una pipeline di esempio

Esempio di semplice pipeline che estrae da csv con Spark

Come pianificare?

Manualmente
Strumento di pianificazione cron
E le dipendenze?

DAG

Directed Acyclic Graph

Insieme di nodi
Archi direzionati
Nessun ciclo

Esempio di DAG

Gli strumenti giusti

cron di Linux
Prefect e Dagster
Apache Airflow

Logo di Apache Airflow

Creato in Airbnb
DAG
Python

Airflow: un DAG di esempio

Esempio di DAG Airflow

Airflow: un esempio in codice

@dag(dag_id="example_dag",
     start_date=datetime(2024, 1, 1),
     schedule="0 * * * *")
def example_dag():

    @task
    def start_cluster(): ...
    @task
    def ingest_customer_data(): ...
    @task
    def ingest_product_data(): ...
    @task
    def enrich_customer_data(): ...

Airflow: un esempio in codice

@dag(dag_id="example_dag", ...)
def example_dag():
    ...
    # Imposta il flusso di dipendenze
    cluster = start_cluster()
    customers = ingest_customer_data()
    products = ingest_product_data()
    cluster >> [customers, products]
    [customers, products] >> enrich_customer_data()
# Esegui il DAG
example_dag()

Esercitiamoci!

Introduzione al Data Engineering