Decyzje, które podejmiesz w pierwszym tygodniu, decydują o tym, czy projekt pipeline'ów będzie rosnąć spokojnie, czy po dwunastu miesiącach runie pod własnym ciężarem
Jest taki rodzaj projektu Airflow, który większość inżynierów danych albo przeżyła, albo o nim słyszała. Zaczyna się czysto: kilka DAGów, jasny cel, mały zespół. Potem zmieniają się wymagania. Pojawiają się nowe źródła danych. Stakeholderzy chcą więcej pipeline'ów, szybciej. Ktoś dokłada workaround tu, skopiowanego DAGa tam. W ciągu roku codebase staje się labiryntem zduplikowanej logiki, niespójnego nazewnictwa, bez monitoringu i z testami, które istnieją tylko w teorii. Deploy zmiany przypomina rozbrajanie bomby.
Najbardziej frustrujące jest to, że większości tego da się uniknąć. Nie przez pisanie idealnego kodu od dnia pierwszego, tylko przez podjęcie kilku strukturalnych decyzji przed napisaniem pierwszego DAGa. Ten artykuł jest właśnie o tych decyzjach.
Wymagania będą się zmieniać szybciej, niż oczekujesz
Każdy projekt Airflow, który widziałem, zaczynał się od założeń, które później okazywały się błędne. Pipeline, który miał działać raz dziennie, musi teraz działać co godzinę. Źródło, które miało być stabilne, zaczyna zmieniać schemat. Zespół, który liczył trzy osoby, urośnie do dziesięciu - i nagle „wszyscy po prostu wiedzą, jak to działa” przestaje być sensowną strategią dokumentacji.
To nie jest porażka planowania. To natura inżynierii danych w organizacjach, które faktycznie korzystają ze swoich danych. Pytanie nie brzmi, jak zapobiec zmianom wymagań. Brzmi: jak zbudować system, który absorbuje zmiany bez przepisywania wszystkiego za każdym razem.
Odpowiedzią jest abstrakcja - i czas na jej wprowadzenie jest przed tym, zanim będziesz mieć dziesięć DAGów, z których każdy robi to samo trochę inaczej.
Najczęstsza pułapka to pisanie DAGów jak samodzielnych skryptów. Każdy DAG definiuje własne połączenia, własną logikę retry, własne powiadomienia. Działa dobrze, dopóki nie trzeba zmienić retry we wszystkich pipeline'ach - i wtedy robisz tę samą zmianę w czterdziestu miejscach.
Zamiast tego potraktuj projekt Airflow jak paczkę Pythona z prawdziwymi praktykami software engineeringu. Wydziel współdzielone zachowanie do klas bazowych i funkcji pomocniczych. Zdefiniuj standardową fabrykę DAGów, która koduje Twoje domyślne ustawienia. Gdy wymagania się zmienią, zmieniasz fabrykę - i każdy DAG, który z niej korzysta, dostaje aktualizację za darmo.
Minimalny przykład, jak to wygląda w praktyce:
# pipelines/factory.py
from airflow import DAG
from datetime import datetime
from pipelines.callbacks import notify_on_failure
DEFAULT_ARGS = {
"owner": owner,
"retries": 2,
"retry_delay": timedelta(minutes=5),
"on_failure_callback": notify_on_failure,
}
def create_dag(
dag_id: str,
schedule: str,
owner: str,
tags: list[str],
**kwargs
) -> DAG:
default_args = kwargs.get('default_args', DEFAULT_ARGS)
return DAG(
dag_id=dag_id,
schedule=schedule,
start_date=datetime(2024, 1, 1),
default_args=default_args,
tags=tags,
catchup=False,
**kwargs,
)
Każdy DAG w projekcie woła create_dag(). Logika retry żyje w jednym miejscu. Callbacki przy failu też. Gdy trzeba dodać query tagging, zmienić domyślną liczbę retry albo wymienić system powiadomień - zmieniasz jeden plik.
Przy dwóch DAGach to wygląda na zbędną abstrakcję. Przy pięćdziesięciu - na ratunek.
SLA
Nikt nie chce na kickoffie projektu rozmawiać o SLA. Wydaje się to przedwczesne, biurokratyczne i oderwane od prawdziwej pracy budowania. Pomiń to - i za pół roku będziesz mieć dużo gorszą rozmowę, gdy stakeholder zapyta, dlaczego pipeline padał w ciszy przez trzy dni i nikt tego nie zauważył.
SLA w kontekście Airflow to nie tylko „jak szybko pipeline musi się wykonać”. To zestaw ustaleń, które napędzają konkretne decyzje inżynierskie. Zanim napiszesz pierwszego DAGa, warto ustalić przynajmniej to:
Jaka jest akceptowalna latencja dla każdego pipeline'u? Pipeline zasilający dashboard w czasie rzeczywistym ma inne wymagania niż ten, który wypełnia tygodniowy raport. To determinuje scheduling, timeouty i to, jak agresywnie trzeba optymalizować.
Jaka jest akceptowalna stopa awarii? Niektóre pipeline'y tolerują okazjonalne failę i nadrabiają w następnym runie. Inne nie mogą pominąć ani jednego wykonania. To determinuje strategię retry, progi alertów i to, czy potrzebujesz logiki backfill.
Kogo trzeba powiadomić, gdy coś się zepsuje - i jak szybko? „Zespół zauważy” to nie jest SLA. Zdefiniuj konkretną ścieżkę eskalacji: które failę idą natychmiast na Slacka, które generują ticket, które wymagają page'a. Ta rozmowa zmusza stakeholderów do realnego priorytetyzowania.
Co liczy się jako naruszenie? DAG, który się wykona, ale zwróci błędny wynik, jest gorszy niż DAG, który pada głośno. Uzgodnienie, co znaczy „działa poprawnie”, przed zbudowaniem pipeline'u jest dużo łatwiejsze niż uzgodnienie tego po incydencie.
Airflow ma natywne SLA miss - możesz ustawić sla na poszczególnych taskach, a Airflow wywoła callback, gdy task przekroczy oczekiwany czas. Warto to podpiąć od początku, bo dokładanie monitoringu SLA do działającego pipeline'u produkcyjnego to jedna z tych rzeczy, które zawsze spadają w priorytetach.
from datetime import timedelta
with create_dag("orders_pipeline", schedule="@hourly", owner="data-eng", tags=["orders"]) as dag:
transform = PythonOperator(
task_id="transform_orders",
python_callable=transform_orders,
sla=timedelta(minutes=30), # alert if this task takes longer than 30 min
)
Wczesne rozmowy pomagają też w subtelnym problemie: zarządzaniu oczekiwaniami. Stakeholderzy, którzy pomogli zdefiniować SLA, rozumieją, dlaczego zdarzył się incydent i co to oznacza. Stakeholderzy, których nikt nie zapytał, zwykle zakładają, że system powinien być idealny.
Monitoring
DAG Airflow, który pada po cichu, jest gorszy niż brak DAGa w ogóle - bo tworzy iluzję, że pipeline działa. W środowisku produkcyjnym monitoring nie jest opcjonalny. Pytanie brzmi tylko, jak wdrożyć go dobrze.
Bazowy wzorzec: podłącz failure callbacki do Slacka i maila na poziomie fabryki DAGów, żeby każdy pipeline dostawał powiadomienia bez konfiguracji per DAG. Callback dostaje z Airflow obiekt context z wszystkim, czego potrzebujesz do sensownego alertu.
# pipelines/callbacks.py
from airflow.hooks.base import BaseHook
from slack_sdk import WebClient
def notify_on_failure(context):
dag_id = context["dag"].dag_id
task_id = context["task_instance"].task_id
execution_date = context["execution_date"]
log_url = context["task_instance"].log_url
client = WebClient(token=BaseHook.get_connection("slack_default").password)
client.chat_postMessage(
channel="#data-alerts",
text=(
f":red_circle: *Pipeline failure* "
f"*DAG:* `{dag_id}` "
f"*Task:* `{task_id}` "
f"*Run:* `{execution_date}` "
f"<{log_url}|View logs>"
),
)
Kilka rzeczy odróżnia monitoring, który naprawdę pomaga, od tego, którego ludzie się nauczyli ignorować. Po pierwsze, każdy alert powinien mieć bezpośredni link do logów. Alert, który wymaga trzech kliknięć, zanim zobaczysz, co poszło nie tak, tworzy tarcia i kończy się odkładaniem alertów na bok. Po drugie, alerty powinny iść według severity i ownershipu. Fail w pipeline'ie zespołu finance powinien iść na kanał finance, nie na generyczny #data-alerts, którego w ciągu dnia nikt nie czyta. Po trzecie, szum z retry to realny problem. Jeśli task zrobi dwa retry i dopiero potem przejdzie, nie chcesz dwóch alertów o failu i jednego o sukcesie. Skonfiguruj callbacki tak, by odpalały się dopiero przy finalnym failu, po wyczerpaniu retry.
Poza alertami o failach warto śledzić czas trwania DAGów w czasie. Pipeline, który kiedyś biegł dziesięć minut, a teraz czterdzieści, nie failuje - ale coś jest nie tak. Airflow udostępnia te dane przez bazę metadanych; warto zbudować prosty dashboard albo okresowy check, który flaguje pipeline'y, których runtime dryfuje w górę.
Testowanie
Testowanie pipeline'ów Airflow jest bardziej zależne od środowiska niż testowanie większości oprogramowania - i udawanie czegoś innego prowadzi do rad, które się nie uogólniają. Zespół na Astronomerze ma inne opcje niż zespół na MWAA albo self-hosted Kubernetesie. Zamiast przepisywać konkretny framework, oto zasady, które trzymają się niezależnie od środowiska.
Testuj Pythona, nie Airflow. Najbardziej wiarygodne testy to te, które sprawdzają logikę biznesową w operatorach i callable'ach w całkowitej izolacji, bez runtime'u Airflow. Jeśli funkcja transform_orders zawiera logikę wartą testowania, napisz do niej unit test. Zamockuj połączenia do bazy. Assercje na output. Te testy są szybkie, stabilne i niezależne od środowiska.
Testuj strukturę DAGa osobno od wykonania DAGa. Airflow daje narzędzia do importu i inspekcji obiektów DAG bez ich uruchamiania. Użyj ich, by asercjami sprawdzić, że DAG ma oczekiwane taski, zależności i konfigurację. To łapie szeroką klasę bugów przed deployem, bez działającego środowiska Airflow.
# tests/test_dag_structure.py
from airflow.models import DagBag
def test_orders_dag_loads():
dagbag = DagBag(dag_folder="dags/", include_examples=False)
assert "orders_pipeline" in dagbag.dags
assert len(dagbag.import_errors) == 0
def test_orders_dag_task_count():
dagbag = DagBag(dag_folder="dags/", include_examples=False)
dag = dagbag.dags["orders_pipeline"]
assert len(dag.tasks) == 3
Bądź szczery wobec testów integracyjnych. Pełne testy end-to-end, które odpalają DAGi na prawdziwej infrastrukturze, są wartościowe - ale też wolne, kruche i drogie w utrzymaniu. W większości zespołów właściwa odpowiedź to niewielka liczba starannie wybranych testów integracyjnych na krytyczne ścieżki, a nie kompletne pokrycie każdego DAGa. Jasno wyznacz granicę, żeby zespół wiedział, co suite gwarantuje, a czego nie.
Linting
Code review łapie bugi, ale nie skaluje się jako mechanizm egzekwowania spójności. Wraz ze wzrostem zespołu „dogadaliśmy konwencje” zamienia się w „jakoś pamiętamy konwencje”, a potem w „każdy robi to trochę inaczej”. Linting automatyzuje egzekucję, żeby konwencje nie zależały od tego, czy reviewer pamięta sprawdzić.
W projektach Airflow, które w rdzeniu są Pythonem, tooling jest dojrzały i wart kosztów setupu.
Ruff to dziś standard lintingu i formatowania Pythona. Jest wyraźnie szybszy niż flake8 czy pylint, pokrywa szeroki zestaw reguł i łączy linting z formatowaniem w jednym narzędziu. W projekcie Airflow kluczowe reguły to m.in. kolejność importów, wykrywanie nieużywanych importów oraz checki złożoności, które flagują DAGi albo callable'e robiące się zbyt duże.
Pre-commit hooks odpalają linting automatycznie przed każdym commitem. Wartość nie leży tylko w wcześniejszym łapaniu problemów - także w tym, że nie trzeba pamiętać o ręcznym odpalaniu lintera. Developer, który raz ustawi pre-commit, więcej o tym nie myśli.
# .pre-commit-config.yaml
repos:
- repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.4.4
hooks:
- id: ruff
args: [--fix]
- id: ruff-format
Szczera uwaga: integracje pre-commit nie zawsze dogadują się z każdym środowiskiem deployu Airflow. Zespoły spotykały konflikty wersji hooków z środowiskiem Pythona, w którym działa Airflow - hooki nie instalowały się albo zachowywały się nieprzewidywalnie. Rozwiązanie: bądź konserwatywny z wersjami hooków, pinuj wszystko explicite i przetestuj pełny setup pre-commit w czystym środowisku, zanim wdrożysz go w zespole. Jeśli konkretny hook sprawia stałe problemy, wyjmij go i odpalaj ten check w CI - linter w CI, ale nie lokalnie, i tak jest dużo lepszy niż linter, który nie działa nigdzie.
Celem nie jest najbardziej wyczerpujący setup lintingu. Celem jest setup stabilny, którego zespół faktycznie używa, i który egzekwuje podzbiór reguł najważniejszych dla Waszego codebase'u.
Składając to razem
Żadna z tych decyzji nie jest sama w sobie skomplikowana. Trudność polega na tym, że wszystkie muszą wydarzyć się, zanim projekt nabierze rozpędu - wtedy, gdy wydaje się, że spowalniacie robotę rzeczami, które da się odłożyć.
Koszt odkładania nie jest oczywisty, dopóki nie jest za późno. Projekt Airflow bez fabryki DAGów to projekt, w którym zmiana domyślnego retry wymaga ruszenia każdego pliku DAGa. Projekt bez zdefiniowanych SLA to projekt, w którym każdy incydent staje się negocjacją o oczekiwaniach. Projekt bez monitoringu to projekt, w którym failę odkrywają użytkownicy wcześniej niż zespół.
Zbuduj strukturę, zanim jej będziesz potrzebować. Pierwszy tydzień to najtańszy moment. I zespoły, które robią to konsekwentnie, to te, które po dwóch latach nadal shipują szybko - podczas gdy wszyscy inni rozmawiają o rewrite.
Joachim Hodana - Software & Data Engineer
Airflow gotowy na enterprise: o czym pomyśleć, zanim napiszesz pierwszego DAGa został pierwotnie opublikowany w Lortech Solutions Blog na Medium, gdzie rozmowa trwa dalej dzięki podświetleniom i odpowiedziom czytelników.
