Szablony, idempotentność i uzupełnianie danych

Budowanie potoków danych z Airflow

Volker Janz

Senior Developer Advocate at Astronomer

Szablony Jinja w Airflow

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

{{ ds }} renderuje datę logiczną w formacie YYYY-MM-DD
Każde uruchomienie ma własną datę, więc to samo zadanie daje różne wyniki
Działa tylko w polach obsługujących szablony (np. bash_command, sql)

Problem idempotentności

Uruchomienie 1 (31 marca):

INSERT danych sprzedaży za 31 marca
Wynik: 3 wiersze

Uruchomienie 2 (ponowne uruchomienie dla 31 marca):

INSERT danych sprzedaży za 31 marca — ponownie
Wynik: 6 wierszy (duplikaty!)

Zduplikowane wiersze po ponownym uruchomieniu

Wzorzec: usuń i wstaw

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,
)

Można też użyć podejścia UPSERT lub MERGE

Usuń-a-następnie-wstaw

Uzupełnianie danych (backfilling)

Ponowne przetwarzanie zakresu historycznych dat
Airflow tworzy jedno uruchomienie DAG-a na datę logiczną
W połączeniu z idempotentsością uzupełnianie danych jest bezpieczne

Uzupełnianie zależy od harmonogramu

Backfill w interfejsie

Wyzwól DAG-a, wybierz Backfill, ustaw daty Od i Do
Możesz uruchomić Brakujące uruchomienia, Brakujące i błędne lub Wszystkie uruchomienia
Poziom równoległości ustawisz przez Max Active Runs, a kolejność przetwarzania możesz zmienić
Run Backfill uruchamia proces i tworzy poszczególne uruchomienia

Interfejs backfill w Airflow

Czas na praktykę!

Budowanie potoków danych z Airflow