Templating, idempotență și backfilling

Construirea pipeline-urilor de date cu Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Template-uri Jinja în Airflow

@task.bash
def export_data():
    return "cp /data/sales.csv /export/sales_{{ ds }}.csv"

{{ ds }} redă data logică în formatul YYYY-MM-DD
Fiecare rulare primește propria dată, astfel că aceeași sarcină produce rezultate diferite per rulare
Funcționează doar în câmpuri cu suport pentru templating (precum bash_command, sql)

Problema idempotenței

Rularea 1 (31 martie):

INSERT pentru vânzările din 31 martie
Rezultat: 3 rânduri

Rularea 2 (re-rulare 31 martie):

INSERT din nou pentru vânzările din 31 martie
Rezultat: 6 rânduri (duplicate!)

Rânduri duplicate la re-rulare

Tiparul delete-then-insert

staged_rows = SQLExecuteQueryOperator(
    task_id="get_staged_rows",
    conn_id="duckdb_default",
    sql="SELECT * FROM staging WHERE date = '{{ ds }}'",
)

load = SQLInsertRowsOperator(
    task_id="load_sales",
    conn_id="duckdb_default",
    table_name="sales",
    columns=["date", "product", "amount"],
    preoperator="DELETE FROM sales WHERE date = '{{ ds }}';",
    rows=staged_rows.output,
)

Se pot folosi și abordările UPSERT sau MERGE

Delete-then-insert

Backfilling

Reprocesează un interval de date istorice
Airflow creează câte o rulare DAG per dată logică
Combinat cu idempotența, backfilling-ul este sigur

Backfill-ul depinde de programare

Backfill în interfața UI

Declanșează un DAG, selectează Backfill, setează datele From și To
Poți alege să declanșezi Missing Runs, Missing and Errored Runs sau All Runs
Poți seta paralelismul cu Max Active Runs și schimba ordinea de execuție
Cu Run Backfill procesul pornește și rulările sunt create

Interfața UI pentru backfill în Airflow

Hai să exersăm!

Construirea pipeline-urilor de date cu Airflow